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/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/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..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 @@ -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 final ReentrantLock bodyBufferLock = 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()); + bodyBufferLock.lock(); + try { + if (bodyBuffer == null) { + bodyBuffer = serializeStruct(toStruct()).array(); + } + } finally { + bodyBufferLock.unlock(); + } + return new NetworkSend(destination, new ByteBuffer[]{headerBuffer, ByteBuffer.wrap(bodyBuffer)}); + } + @Override protected Struct toStruct() { structLock.lock(); diff --git a/core/src/main/scala/kafka/controller/ControllerChannelManager.scala b/core/src/main/scala/kafka/controller/ControllerChannelManager.scala index 3990f4621964d..c4d5b861c466e 100755 --- a/core/src/main/scala/kafka/controller/ControllerChannelManager.scala +++ b/core/src/main/scala/kafka/controller/ControllerChannelManager.scala @@ -524,14 +524,26 @@ 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 => + 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() }