[LI-HOTFIX] LIKAFKA-18349 - Reuse the same UpdateMetadataRequest object and struct to reduce controller memory usage

TICKET = KAFKA-7186
LI_DESCRIPTION =

During controller failover and broker changes, it sends out UpdateMetadataRequest to all brokers in the cluster containing the states for all partitions and live brokers. The current implementation will instantiate the UpdateMetadataRequest object and its serialized form (Struct) for # of brokers times, which causes OOM if the memory exceeds the configure JVM heap size before GC kicks in. We have seen this issue in the production environment for multiple times.
For example, if we have 100 brokers in the cluster and each broker is the leader of 2k partitions, the extra memory usage introduced by controller trying to send out UpdateMetadataRequest is around:
```
<memory used by UpdateMetadataRequest Structs> * <# of brokers> * <total # of leader parittions> = 250B * 100 * 200k = 5GB
```
This patch avoids the recreation of UpdateMetadataReuqest and struct objects if the same builder object is used to build the request. This can significantly reduce memory footprint in Controller based on the above calculation.


RB=1364716
BUG=LIKAFKA-18349
G=Kafka-Code-Reviews
R=luwang,okaraman,mgharat
A=mgharat

EXIT_CRITERIA = TICKET [KAFKA-7186]

Author: hzxa21

clients/src/main/java/org/apache/kafka/common/requests/UpdateMetadataRequest.java

@@ -185,6 +185,9 @@ public static class Builder extends AbstractControlRequest.Builder<UpdateMetadat
         private final Map<TopicPartition, PartitionState> partitionStates;
         private final Set<Broker> liveBrokers;
 
+        // LIKAFKA-18349 - Cache the UpdateMetadataRequest Objects to reduce memory usage
+        private final Map<Short, UpdateMetadataRequest> requestCache = new HashMap<>();
+
         public Builder(short version, int controllerId, int controllerEpoch, long brokerEpoch,
                        Map<TopicPartition, PartitionState> partitionStates, Set<Broker> liveBrokers) {
             super(ApiKeys.UPDATE_METADATA, version, controllerId, controllerEpoch, brokerEpoch);
@@ -194,14 +197,20 @@ public Builder(short version, int controllerId, int controllerEpoch, long broker
 
         @Override
         public UpdateMetadataRequest build(short version) {
-            if (version == 0) {
-                for (Broker broker : liveBrokers) {
-                    if (broker.endPoints.size() != 1 || broker.endPoints.get(0).securityProtocol != SecurityProtocol.PLAINTEXT) {
-                        throw new UnsupportedVersionException("UpdateMetadataRequest v0 only handles PLAINTEXT endpoints");
+            UpdateMetadataRequest updateMetadataRequest = requestCache.get(version);
+            if (updateMetadataRequest == null) {
+                if (version == 0) {
+                    for (Broker broker : liveBrokers) {
+                        if (broker.endPoints.size() != 1 || broker.endPoints.get(0).securityProtocol != SecurityProtocol.PLAINTEXT) {
+                            throw new UnsupportedVersionException(
+                                "UpdateMetadataRequest v0 only handles PLAINTEXT endpoints");
+                        }
                     }
                 }
+                updateMetadataRequest = new UpdateMetadataRequest(version, controllerId, controllerEpoch, brokerEpoch, partitionStates, liveBrokers);
+                requestCache.put(version, updateMetadataRequest);
             }
-            return new UpdateMetadataRequest(version, controllerId, controllerEpoch, brokerEpoch, partitionStates, liveBrokers);
+            return updateMetadataRequest;
         }
 
         @Override
@@ -327,6 +336,9 @@ public String toString() {
     private final Map<TopicPartition, PartitionState> partitionStates;
     private final Set<Broker> liveBrokers;
 
+    // LIKAFKA-18349 - Cache the UpdateMetadataRequest struct to reduce memory usage
+    private Struct struct = null;
+
     private UpdateMetadataRequest(short version, int controllerId, int controllerEpoch, long brokerEpoch,
                                   Map<TopicPartition, PartitionState> partitionStates, Set<Broker> liveBrokers) {
         super(ApiKeys.UPDATE_METADATA, version, controllerId, controllerEpoch, brokerEpoch);
@@ -400,74 +412,74 @@ public UpdateMetadataRequest(Struct struct, short versionId) {
 
     @Override
     protected Struct toStruct() {
-        short version = version();
-        Struct struct = new Struct(ApiKeys.UPDATE_METADATA.requestSchema(version));
-        struct.set(CONTROLLER_ID, controllerId);
-        struct.set(CONTROLLER_EPOCH, controllerEpoch);
-        struct.setIfExists(BROKER_EPOCH, brokerEpoch);
-
-        if (struct.hasField(TOPIC_STATES)) {
-            Map<String, Map<Integer, PartitionState>> topicStates = CollectionUtils.groupPartitionDataByTopic(partitionStates);
-            List<Struct> topicStatesData = new ArrayList<>(topicStates.size());
-            for (Map.Entry<String, Map<Integer, PartitionState>> entry : topicStates.entrySet()) {
-                Struct topicStateData = struct.instance(TOPIC_STATES);
-                topicStateData.set(TOPIC_NAME, entry.getKey());
-                Map<Integer, PartitionState> partitionMap = entry.getValue();
-                List<Struct> partitionStatesData = new ArrayList<>(partitionMap.size());
-                for (Map.Entry<Integer, PartitionState> partitionEntry : partitionMap.entrySet()) {
-                    Struct partitionStateData = topicStateData.instance(PARTITION_STATES);
-                    partitionStateData.set(PARTITION_ID, partitionEntry.getKey());
-                    partitionEntry.getValue().setStruct(partitionStateData);
+        if (struct == null) {
+            short version = version();
+            Struct struct = new Struct(ApiKeys.UPDATE_METADATA.requestSchema(version));
+            struct.set(CONTROLLER_ID, controllerId);
+            struct.set(CONTROLLER_EPOCH, controllerEpoch);
+            struct.setIfExists(BROKER_EPOCH, brokerEpoch);
+
+            if (struct.hasField(TOPIC_STATES)) {
+                Map<String, Map<Integer, PartitionState>> topicStates = CollectionUtils.groupPartitionDataByTopic(partitionStates);
+                List<Struct> topicStatesData = new ArrayList<>(topicStates.size());
+                for (Map.Entry<String, Map<Integer, PartitionState>> entry : topicStates.entrySet()) {
+                    Struct topicStateData = struct.instance(TOPIC_STATES);
+                    topicStateData.set(TOPIC_NAME, entry.getKey());
+                    Map<Integer, PartitionState> partitionMap = entry.getValue();
+                    List<Struct> partitionStatesData = new ArrayList<>(partitionMap.size());
+                    for (Map.Entry<Integer, PartitionState> partitionEntry : partitionMap.entrySet()) {
+                        Struct partitionStateData = topicStateData.instance(PARTITION_STATES);
+                        partitionStateData.set(PARTITION_ID, partitionEntry.getKey());
+                        partitionEntry.getValue().setStruct(partitionStateData);
+                        partitionStatesData.add(partitionStateData);
+                    }
+                    topicStateData.set(PARTITION_STATES, partitionStatesData.toArray());
+                    topicStatesData.add(topicStateData);
+                }
+                struct.set(TOPIC_STATES, topicStatesData.toArray());
+            } else {
+                List<Struct> partitionStatesData = new ArrayList<>(partitionStates.size());
+                for (Map.Entry<TopicPartition, PartitionState> entry : partitionStates.entrySet()) {
+                    Struct partitionStateData = struct.instance(PARTITION_STATES);
+                    TopicPartition topicPartition = entry.getKey();
+                    partitionStateData.set(TOPIC_NAME, topicPartition.topic());
+                    partitionStateData.set(PARTITION_ID, topicPartition.partition());
+                    entry.getValue().setStruct(partitionStateData);
                     partitionStatesData.add(partitionStateData);
                 }
-                topicStateData.set(PARTITION_STATES, partitionStatesData.toArray());
-                topicStatesData.add(topicStateData);
+                struct.set(PARTITION_STATES, partitionStatesData.toArray());
             }
-            struct.set(TOPIC_STATES, topicStatesData.toArray());
-        } else {
-            List<Struct> partitionStatesData = new ArrayList<>(partitionStates.size());
-            for (Map.Entry<TopicPartition, PartitionState> entry : partitionStates.entrySet()) {
-                Struct partitionStateData = struct.instance(PARTITION_STATES);
-                TopicPartition topicPartition = entry.getKey();
-                partitionStateData.set(TOPIC_NAME, topicPartition.topic());
-                partitionStateData.set(PARTITION_ID, topicPartition.partition());
-                entry.getValue().setStruct(partitionStateData);
-                partitionStatesData.add(partitionStateData);
-            }
-            struct.set(PARTITION_STATES, partitionStatesData.toArray());
-        }
 
-        List<Struct> brokersData = new ArrayList<>(liveBrokers.size());
-        for (Broker broker : liveBrokers) {
-            Struct brokerData = struct.instance(LIVE_BROKERS);
-            brokerData.set(BROKER_ID, broker.id);
-
-            if (version == 0) {
-                EndPoint endPoint = broker.endPoints.get(0);
-                brokerData.set(HOST, endPoint.host);
-                brokerData.set(PORT, endPoint.port);
-            } else {
-                List<Struct> endPointsData = new ArrayList<>(broker.endPoints.size());
-                for (EndPoint endPoint : broker.endPoints) {
-                    Struct endPointData = brokerData.instance(ENDPOINTS);
-                    endPointData.set(PORT, endPoint.port);
-                    endPointData.set(HOST, endPoint.host);
-                    endPointData.set(SECURITY_PROTOCOL_TYPE, endPoint.securityProtocol.id);
-                    if (version >= 3)
-                        endPointData.set(LISTENER_NAME, endPoint.listenerName.value());
-                    endPointsData.add(endPointData);
-
-                }
-                brokerData.set(ENDPOINTS, endPointsData.toArray());
-                if (version >= 2) {
-                    brokerData.set(RACK, broker.rack);
+            List<Struct> brokersData = new ArrayList<>(liveBrokers.size());
+            for (Broker broker : liveBrokers) {
+                Struct brokerData = struct.instance(LIVE_BROKERS);
+                brokerData.set(BROKER_ID, broker.id);
+
+                if (version == 0) {
+                    EndPoint endPoint = broker.endPoints.get(0);
+                    brokerData.set(HOST, endPoint.host);
+                    brokerData.set(PORT, endPoint.port);
+                } else {
+                    List<Struct> endPointsData = new ArrayList<>(broker.endPoints.size());
+                    for (EndPoint endPoint : broker.endPoints) {
+                        Struct endPointData = brokerData.instance(ENDPOINTS);
+                        endPointData.set(PORT, endPoint.port);
+                        endPointData.set(HOST, endPoint.host);
+                        endPointData.set(SECURITY_PROTOCOL_TYPE, endPoint.securityProtocol.id);
+                        if (version >= 3) endPointData.set(LISTENER_NAME, endPoint.listenerName.value());
+                        endPointsData.add(endPointData);
+                    }
+                    brokerData.set(ENDPOINTS, endPointsData.toArray());
+                    if (version >= 2) {
+                        brokerData.set(RACK, broker.rack);
+                    }
                 }
-            }
 
-            brokersData.add(brokerData);
+                brokersData.add(brokerData);
+            }
+            struct.set(LIVE_BROKERS, brokersData.toArray());
+            this.struct = struct;
         }
-        struct.set(LIVE_BROKERS, brokersData.toArray());
-
         return struct;
     }