Cache the UpdateMetadataRequest when using the latest inter broker protocol version

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);

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;
+    }
 }

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();

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()
   }