From 1b26550799bf4df65336afb98481e26825cb52ce Mon Sep 17 00:00:00 2001 From: Lucas Wang Date: Mon, 9 Mar 2020 12:33:25 -0700 Subject: [PATCH 1/6] Cache the UpdateMetadataRequest when using the latest inter broker protocol version --- .../kafka/common/network/NetworkSend.java | 16 +++++++++++++ .../common/requests/AbstractRequest.java | 10 ++++++-- .../requests/AbstractRequestResponse.java | 7 ++++++ .../controller/ControllerChannelManager.scala | 23 +++++++++++++++---- 4 files changed, 49 insertions(+), 7 deletions(-) diff --git a/clients/src/main/java/org/apache/kafka/common/network/NetworkSend.java b/clients/src/main/java/org/apache/kafka/common/network/NetworkSend.java index 8820059aa7985..43739c4003e1c 100644 --- a/clients/src/main/java/org/apache/kafka/common/network/NetworkSend.java +++ b/clients/src/main/java/org/apache/kafka/common/network/NetworkSend.java @@ -27,10 +27,26 @@ public NetworkSend(String destination, ByteBuffer buffer) { super(destination, sizeDelimit(buffer)); } + public NetworkSend(String destination, ByteBuffer[] buffers) { + super(destination, sizeDelimit(buffers)); + } + private static ByteBuffer[] sizeDelimit(ByteBuffer buffer) { return new ByteBuffer[] {sizeBuffer(buffer.remaining()), buffer}; } + private static ByteBuffer[] sizeDelimit(ByteBuffer[] buffers) { + int totalRemaining = 0; + for (ByteBuffer byteBuffer: buffers) { + totalRemaining += byteBuffer.remaining(); + } + + ByteBuffer[] result = new ByteBuffer[buffers.length + 1]; + result[0] = sizeBuffer(totalRemaining); + System.arraycopy(buffers, 0, result, 1, buffers.length); + return result; + } + private static ByteBuffer sizeBuffer(int size) { ByteBuffer sizeBuffer = ByteBuffer.allocate(4); sizeBuffer.putInt(size); diff --git a/clients/src/main/java/org/apache/kafka/common/requests/AbstractRequest.java b/clients/src/main/java/org/apache/kafka/common/requests/AbstractRequest.java index c199f8ee5a066..5db8935009098 100644 --- a/clients/src/main/java/org/apache/kafka/common/requests/AbstractRequest.java +++ b/clients/src/main/java/org/apache/kafka/common/requests/AbstractRequest.java @@ -27,7 +27,7 @@ import java.util.Map; public abstract class AbstractRequest extends AbstractRequestResponse { - + private byte[] bodyBuffer; public static abstract class Builder { private final ApiKeys apiKey; private final short oldestAllowedVersion; @@ -93,7 +93,13 @@ public short version() { } public Send toSend(String destination, RequestHeader header) { - return new NetworkSend(destination, serialize(header)); + // For UpdateMetadataRequest, the toSend method on the same object will be called many times, each time with a different destination + // value and a header containing a different correlation id. + ByteBuffer headerBuffer = serializeStruct(header.toStruct()); + if (bodyBuffer == null) { + bodyBuffer = serializeStruct(toStruct()).array(); + } + return new NetworkSend(destination, new ByteBuffer[]{headerBuffer, ByteBuffer.wrap(bodyBuffer)}); } /** diff --git a/clients/src/main/java/org/apache/kafka/common/requests/AbstractRequestResponse.java b/clients/src/main/java/org/apache/kafka/common/requests/AbstractRequestResponse.java index 0ba373d6fea7a..0356b11873f82 100644 --- a/clients/src/main/java/org/apache/kafka/common/requests/AbstractRequestResponse.java +++ b/clients/src/main/java/org/apache/kafka/common/requests/AbstractRequestResponse.java @@ -31,4 +31,11 @@ public static ByteBuffer serialize(Struct headerStruct, Struct bodyStruct) { buffer.rewind(); return buffer; } + + public static ByteBuffer serializeStruct(Struct struct) { + ByteBuffer buffer = ByteBuffer.allocate(struct.sizeOf()); + struct.writeTo(buffer); + buffer.rewind(); + return buffer; + } } diff --git a/core/src/main/scala/kafka/controller/ControllerChannelManager.scala b/core/src/main/scala/kafka/controller/ControllerChannelManager.scala index 3990f4621964d..0f36cfc0da914 100755 --- a/core/src/main/scala/kafka/controller/ControllerChannelManager.scala +++ b/core/src/main/scala/kafka/controller/ControllerChannelManager.scala @@ -524,14 +524,27 @@ abstract class AbstractControllerBrokerRequestBatch(config: KafkaConfig, } } - val maxBrokerEpoch = controllerContext.maxBrokerEpoch - updateMetadataRequestBrokerSet.intersect(controllerContext.liveOrShuttingDownBrokerIds).foreach { broker => - val brokerEpoch = controllerContext.liveBrokerIdAndEpochs(broker) + if (updateMetadataRequestVersion >= 6) { + // We should only create one copy UpdateMetadataRequest that should apply to all brokers. + // The goal is to reduce memory footprint on the controller. + val maxBrokerEpoch = controllerContext.maxBrokerEpoch val updateMetadataRequest = new UpdateMetadataRequest.Builder(updateMetadataRequestVersion, controllerId, controllerEpoch, - brokerEpoch, maxBrokerEpoch, partitionStates.asJava, liveBrokers.asJava) - sendRequest(broker, updateMetadataRequest) + AbstractControlRequest.UNKNOWN_BROKER_EPOCH, maxBrokerEpoch, partitionStates.asJava, liveBrokers.asJava) + + updateMetadataRequestBrokerSet.intersect(controllerContext.liveOrShuttingDownBrokerIds).foreach { broker => + val brokerEpoch = controllerContext.liveBrokerIdAndEpochs(broker) + sendRequest(broker, updateMetadataRequest) + } + } else { + updateMetadataRequestBrokerSet.intersect(controllerContext.liveOrShuttingDownBrokerIds).foreach { broker => + val brokerEpoch = controllerContext.liveBrokerIdAndEpochs(broker) + val updateMetadataRequest = new UpdateMetadataRequest.Builder(updateMetadataRequestVersion, controllerId, controllerEpoch, + brokerEpoch, AbstractControlRequest.UNKNOWN_BROKER_EPOCH, partitionStates.asJava, liveBrokers.asJava) + sendRequest(broker, updateMetadataRequest) + } } + updateMetadataRequestBrokerSet.clear() updateMetadataRequestPartitionInfoMap.clear() } From 323f103ba37b30d6f7ef76e56e1097b7dd399ad7 Mon Sep 17 00:00:00 2001 From: Lucas Wang Date: Mon, 9 Mar 2020 12:37:45 -0700 Subject: [PATCH 2/6] Synchronizing access to the body buffer --- .../apache/kafka/common/requests/AbstractRequest.java | 11 +++++++++-- 1 file changed, 9 insertions(+), 2 deletions(-) diff --git a/clients/src/main/java/org/apache/kafka/common/requests/AbstractRequest.java b/clients/src/main/java/org/apache/kafka/common/requests/AbstractRequest.java index 5db8935009098..7fdb9738078b9 100644 --- a/clients/src/main/java/org/apache/kafka/common/requests/AbstractRequest.java +++ b/clients/src/main/java/org/apache/kafka/common/requests/AbstractRequest.java @@ -25,8 +25,10 @@ import java.nio.ByteBuffer; import java.util.Map; +import java.util.concurrent.locks.ReentrantLock; public abstract class AbstractRequest extends AbstractRequestResponse { + private ReentrantLock bobyBufferLock = new ReentrantLock(); private byte[] bodyBuffer; public static abstract class Builder { private final ApiKeys apiKey; @@ -96,8 +98,13 @@ public Send toSend(String destination, RequestHeader header) { // For UpdateMetadataRequest, the toSend method on the same object will be called many times, each time with a different destination // value and a header containing a different correlation id. ByteBuffer headerBuffer = serializeStruct(header.toStruct()); - if (bodyBuffer == null) { - bodyBuffer = serializeStruct(toStruct()).array(); + bobyBufferLock.lock(); + try { + if (bodyBuffer == null) { + bodyBuffer = serializeStruct(toStruct()).array(); + } + } finally { + bobyBufferLock.unlock(); } return new NetworkSend(destination, new ByteBuffer[]{headerBuffer, ByteBuffer.wrap(bodyBuffer)}); } From 55495748adcad7ca001e28485a4b5762665517eb Mon Sep 17 00:00:00 2001 From: Lucas Wang Date: Mon, 9 Mar 2020 12:42:32 -0700 Subject: [PATCH 3/6] Removing unused line --- .../main/scala/kafka/controller/ControllerChannelManager.scala | 1 - 1 file changed, 1 deletion(-) diff --git a/core/src/main/scala/kafka/controller/ControllerChannelManager.scala b/core/src/main/scala/kafka/controller/ControllerChannelManager.scala index 0f36cfc0da914..c4d5b861c466e 100755 --- a/core/src/main/scala/kafka/controller/ControllerChannelManager.scala +++ b/core/src/main/scala/kafka/controller/ControllerChannelManager.scala @@ -532,7 +532,6 @@ abstract class AbstractControllerBrokerRequestBatch(config: KafkaConfig, AbstractControlRequest.UNKNOWN_BROKER_EPOCH, maxBrokerEpoch, partitionStates.asJava, liveBrokers.asJava) updateMetadataRequestBrokerSet.intersect(controllerContext.liveOrShuttingDownBrokerIds).foreach { broker => - val brokerEpoch = controllerContext.liveBrokerIdAndEpochs(broker) sendRequest(broker, updateMetadataRequest) } } else { From c758933b0ec9276dafd514df55cb9be867698838 Mon Sep 17 00:00:00 2001 From: Lucas Wang Date: Wed, 18 Mar 2020 12:19:16 -0700 Subject: [PATCH 4/6] Moved the caching logic to only the UpdateMetadataRequest --- .../common/requests/AbstractRequest.java | 17 ++------------- .../requests/UpdateMetadataRequest.java | 21 +++++++++++++++++++ 2 files changed, 23 insertions(+), 15 deletions(-) diff --git a/clients/src/main/java/org/apache/kafka/common/requests/AbstractRequest.java b/clients/src/main/java/org/apache/kafka/common/requests/AbstractRequest.java index 7fdb9738078b9..c199f8ee5a066 100644 --- a/clients/src/main/java/org/apache/kafka/common/requests/AbstractRequest.java +++ b/clients/src/main/java/org/apache/kafka/common/requests/AbstractRequest.java @@ -25,11 +25,9 @@ import java.nio.ByteBuffer; import java.util.Map; -import java.util.concurrent.locks.ReentrantLock; public abstract class AbstractRequest extends AbstractRequestResponse { - private ReentrantLock bobyBufferLock = new ReentrantLock(); - private byte[] bodyBuffer; + public static abstract class Builder { private final ApiKeys apiKey; private final short oldestAllowedVersion; @@ -95,18 +93,7 @@ public short version() { } public Send toSend(String destination, RequestHeader header) { - // For UpdateMetadataRequest, the toSend method on the same object will be called many times, each time with a different destination - // value and a header containing a different correlation id. - ByteBuffer headerBuffer = serializeStruct(header.toStruct()); - bobyBufferLock.lock(); - try { - if (bodyBuffer == null) { - bodyBuffer = serializeStruct(toStruct()).array(); - } - } finally { - bobyBufferLock.unlock(); - } - return new NetworkSend(destination, new ByteBuffer[]{headerBuffer, ByteBuffer.wrap(bodyBuffer)}); + return new NetworkSend(destination, serialize(header)); } /** diff --git a/clients/src/main/java/org/apache/kafka/common/requests/UpdateMetadataRequest.java b/clients/src/main/java/org/apache/kafka/common/requests/UpdateMetadataRequest.java index 1dbdd0622d27d..677d419da995c 100644 --- a/clients/src/main/java/org/apache/kafka/common/requests/UpdateMetadataRequest.java +++ b/clients/src/main/java/org/apache/kafka/common/requests/UpdateMetadataRequest.java @@ -22,6 +22,8 @@ import org.apache.kafka.common.TopicPartition; import org.apache.kafka.common.errors.UnsupportedVersionException; import org.apache.kafka.common.network.ListenerName; +import org.apache.kafka.common.network.NetworkSend; +import org.apache.kafka.common.network.Send; import org.apache.kafka.common.protocol.ApiKeys; import org.apache.kafka.common.protocol.Errors; import org.apache.kafka.common.protocol.types.Field; @@ -45,6 +47,9 @@ import static org.apache.kafka.common.protocol.types.Type.INT32; public class UpdateMetadataRequest extends AbstractControlRequest { + private ReentrantLock bobyBufferLock = new ReentrantLock(); + private byte[] bodyBuffer; + private static final Field.ComplexArray TOPIC_STATES = new Field.ComplexArray("topic_states", "Topic states"); private static final Field.ComplexArray PARTITION_STATES = new Field.ComplexArray("partition_states", "Partition states"); private static final Field.ComplexArray LIVE_BROKERS = new Field.ComplexArray("live_brokers", "Live brokers"); @@ -447,6 +452,22 @@ public UpdateMetadataRequest(Struct struct, short versionId) { this.liveBrokers = liveBrokers; } + @Override + public Send toSend(String destination, RequestHeader header) { + // For UpdateMetadataRequest, the toSend method on the same object will be called many times, each time with a different destination + // value and a header containing a different correlation id. + ByteBuffer headerBuffer = serializeStruct(header.toStruct()); + bobyBufferLock.lock(); + try { + if (bodyBuffer == null) { + bodyBuffer = serializeStruct(toStruct()).array(); + } + } finally { + bobyBufferLock.unlock(); + } + return new NetworkSend(destination, new ByteBuffer[]{headerBuffer, ByteBuffer.wrap(bodyBuffer)}); + } + @Override protected Struct toStruct() { structLock.lock(); From 9378628cf07874bc4380068bbfd7a603b2c20620 Mon Sep 17 00:00:00 2001 From: Lucas Wang Date: Wed, 18 Mar 2020 12:29:46 -0700 Subject: [PATCH 5/6] Fixing indentation --- .../org/apache/kafka/common/requests/UpdateMetadataRequest.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/clients/src/main/java/org/apache/kafka/common/requests/UpdateMetadataRequest.java b/clients/src/main/java/org/apache/kafka/common/requests/UpdateMetadataRequest.java index 677d419da995c..c4831353380b4 100644 --- a/clients/src/main/java/org/apache/kafka/common/requests/UpdateMetadataRequest.java +++ b/clients/src/main/java/org/apache/kafka/common/requests/UpdateMetadataRequest.java @@ -465,7 +465,7 @@ public Send toSend(String destination, RequestHeader header) { } finally { bobyBufferLock.unlock(); } - return new NetworkSend(destination, new ByteBuffer[]{headerBuffer, ByteBuffer.wrap(bodyBuffer)}); + return new NetworkSend(destination, new ByteBuffer[]{headerBuffer, ByteBuffer.wrap(bodyBuffer)}); } @Override From 94b1a8e58d17bf6befccde06d80fb801a9eb35bd Mon Sep 17 00:00:00 2001 From: Lucas Wang Date: Fri, 20 Mar 2020 17:23:04 -0700 Subject: [PATCH 6/6] Addressing comments --- .../apache/kafka/common/requests/UpdateMetadataRequest.java | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/clients/src/main/java/org/apache/kafka/common/requests/UpdateMetadataRequest.java b/clients/src/main/java/org/apache/kafka/common/requests/UpdateMetadataRequest.java index c4831353380b4..0c4ee91ffa1d3 100644 --- a/clients/src/main/java/org/apache/kafka/common/requests/UpdateMetadataRequest.java +++ b/clients/src/main/java/org/apache/kafka/common/requests/UpdateMetadataRequest.java @@ -47,7 +47,7 @@ import static org.apache.kafka.common.protocol.types.Type.INT32; public class UpdateMetadataRequest extends AbstractControlRequest { - private ReentrantLock bobyBufferLock = new ReentrantLock(); + private final ReentrantLock bodyBufferLock = new ReentrantLock(); private byte[] bodyBuffer; private static final Field.ComplexArray TOPIC_STATES = new Field.ComplexArray("topic_states", "Topic states"); @@ -457,13 +457,13 @@ public Send toSend(String destination, RequestHeader header) { // For UpdateMetadataRequest, the toSend method on the same object will be called many times, each time with a different destination // value and a header containing a different correlation id. ByteBuffer headerBuffer = serializeStruct(header.toStruct()); - bobyBufferLock.lock(); + bodyBufferLock.lock(); try { if (bodyBuffer == null) { bodyBuffer = serializeStruct(toStruct()).array(); } } finally { - bobyBufferLock.unlock(); + bodyBufferLock.unlock(); } return new NetworkSend(destination, new ByteBuffer[]{headerBuffer, ByteBuffer.wrap(bodyBuffer)}); }