Fetch supported produce version from apiVersions.

Author: OmniaGM

Diff for: clients/src/main/java/org/apache/kafka/clients/ApiVersions.java

@@ -34,15 +34,15 @@ public class ApiVersions {
 
     private final Map<String, NodeApiVersions> nodeApiVersions = new HashMap<>();
     private byte maxUsableProduceMagic = RecordBatch.CURRENT_MAGIC_VALUE;
-    private short maxProduceSupportedVersion = ApiKeys.PRODUCE.latestVersion();
+    private short maxSupportedProduceVersion = ApiKeys.PRODUCE.latestVersion();
 
     public synchronized void update(String nodeId, NodeApiVersions nodeApiVersions) {
         this.nodeApiVersions.put(nodeId, nodeApiVersions);
         this.maxUsableProduceMagic = computeMaxUsableProduceMagic();
-        this.maxProduceSupportedVersion = computeMaxProduceSupportedVersion();
+        this.maxSupportedProduceVersion = computeMaxSupportedProduceVersion();
     }
 
-    private short computeMaxProduceSupportedVersion() {
+    private short computeMaxSupportedProduceVersion() {
         Optional<Short> knownBrokerNodesMinSupportedVersionForProduce = this.nodeApiVersions.values().stream()
                 .filter(versions -> versions.apiVersion(ApiKeys.PRODUCE) != null) // filter out Raft controller nodes
                 .map(versions -> versions.latestUsableVersion(ApiKeys.PRODUCE))
@@ -54,6 +54,7 @@ private short computeMaxProduceSupportedVersion() {
     public synchronized void remove(String nodeId) {
         this.nodeApiVersions.remove(nodeId);
         this.maxUsableProduceMagic = computeMaxUsableProduceMagic();
+        this.maxSupportedProduceVersion = computeMaxSupportedProduceVersion();
     }
 
     public synchronized NodeApiVersions get(String nodeId) {
@@ -74,8 +75,8 @@ private byte computeMaxUsableProduceMagic() {
     public synchronized byte maxUsableProduceMagic() {
         return maxUsableProduceMagic;
     }
-    public synchronized short getMaxSupportedProduceVersion() {
-        return maxProduceSupportedVersion;
+    public synchronized short maxSupportedProduceVersion() {
+        return maxSupportedProduceVersion;
     }
 
 }

Diff for: clients/src/main/java/org/apache/kafka/clients/producer/internals/Sender.java

@@ -27,7 +27,6 @@
 import org.apache.kafka.clients.Metadata;
 import org.apache.kafka.clients.MetadataSnapshot;
 import org.apache.kafka.clients.NetworkClientUtils;
-import org.apache.kafka.clients.NodeApiVersions;
 import org.apache.kafka.clients.RequestCompletionHandler;
 import org.apache.kafka.common.InvalidRecordException;
 import org.apache.kafka.common.KafkaException;
@@ -51,7 +50,6 @@
 import org.apache.kafka.common.metrics.stats.Avg;
 import org.apache.kafka.common.metrics.stats.Max;
 import org.apache.kafka.common.metrics.stats.Meter;
-import org.apache.kafka.common.protocol.ApiKeys;
 import org.apache.kafka.common.protocol.Errors;
 import org.apache.kafka.common.record.MemoryRecords;
 import org.apache.kafka.common.record.RecordBatch;
@@ -884,7 +882,7 @@ private void sendProduceRequest(long now, int destination, short acks, int timeo
                 minUsedMagic = batch.magic();
         }
         Map<String, Uuid> topicIds = getTopicIdsFromBatches(batches);
-        boolean canUseTopicId = apiVersions.getMaxSupportedProduceVersion() >= 12;
+        boolean canUseTopicId = apiVersions.maxSupportedProduceVersion() >= 12;
 
         ProduceRequestData.TopicProduceDataCollection tpd = new ProduceRequestData.TopicProduceDataCollection();
         for (ProducerBatch batch : batches) {