MINOR: Remove unused ApiVersions variable from Sender and RecordAccumulator (#19399)

Remove unused `ApiVersions` variable from Sender and RecordAccumulator.

Reviewers: PoAn Yang <[email protected]>, Ken Huang
<[email protected]>, Parker Chang <[email protected]>,
Chia-Ping Tsai <[email protected]>

Author: Yunyung

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

@@ -435,7 +435,6 @@ public KafkaProducer(Properties properties, Serializer<K> keySerializer, Seriali
                     metrics,
                     PRODUCER_METRIC_GROUP_NAME,
                     time,
-                    apiVersions,
                     transactionManager,
                     new BufferPool(this.totalMemorySize, batchSize, metrics, time, PRODUCER_METRIC_GROUP_NAME));
 
@@ -538,8 +537,7 @@ Sender newSender(LogContext logContext, KafkaClient kafkaClient, ProducerMetadat
                 time,
                 requestTimeoutMs,
                 producerConfig.getLong(ProducerConfig.RETRY_BACKOFF_MS_CONFIG),
-                this.transactionManager,
-                apiVersions);
+                this.transactionManager);
     }
 
     private static Compression configureCompression(ProducerConfig config) {

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

@@ -16,7 +16,6 @@
  */
 package org.apache.kafka.clients.producer.internals;
 
-import org.apache.kafka.clients.ApiVersions;
 import org.apache.kafka.clients.CommonClientConfigs;
 import org.apache.kafka.clients.MetadataSnapshot;
 import org.apache.kafka.clients.producer.Callback;
@@ -82,7 +81,6 @@ public class RecordAccumulator {
     private final boolean enableAdaptivePartitioning;
     private final BufferPool free;
     private final Time time;
-    private final ApiVersions apiVersions;
     private final ConcurrentMap<String /*topic*/, TopicInfo> topicInfoMap = new CopyOnWriteMap<>();
     private final ConcurrentMap<Integer /*nodeId*/, NodeLatencyStats> nodeStats = new CopyOnWriteMap<>();
     private final IncompleteBatches incomplete;
@@ -109,7 +107,6 @@ public class RecordAccumulator {
      * @param metrics The metrics
      * @param metricGrpName The metric group name
      * @param time The time instance to use
-     * @param apiVersions Request API versions for current connected brokers
      * @param transactionManager The shared transaction state object which tracks producer IDs, epochs, and sequence
      *                           numbers per partition.
      * @param bufferPool The buffer pool
@@ -125,7 +122,6 @@ public RecordAccumulator(LogContext logContext,
                              Metrics metrics,
                              String metricGrpName,
                              Time time,
-                             ApiVersions apiVersions,
                              TransactionManager transactionManager,
                              BufferPool bufferPool) {
         this.logContext = logContext;
@@ -147,7 +143,6 @@ public RecordAccumulator(LogContext logContext,
         this.incomplete = new IncompleteBatches();
         this.muted = new HashSet<>();
         this.time = time;
-        this.apiVersions = apiVersions;
         nodesDrainIndex = new HashMap<>();
         this.transactionManager = transactionManager;
         registerMetrics(metrics, metricGrpName);
@@ -169,7 +164,6 @@ public RecordAccumulator(LogContext logContext,
      * @param metrics The metrics
      * @param metricGrpName The metric group name
      * @param time The time instance to use
-     * @param apiVersions Request API versions for current connected brokers
      * @param transactionManager The shared transaction state object which tracks producer IDs, epochs, and sequence
      *                           numbers per partition.
      * @param bufferPool The buffer pool
@@ -184,7 +178,6 @@ public RecordAccumulator(LogContext logContext,
                              Metrics metrics,
                              String metricGrpName,
                              Time time,
-                             ApiVersions apiVersions,
                              TransactionManager transactionManager,
                              BufferPool bufferPool) {
         this(logContext,
@@ -198,7 +191,6 @@ public RecordAccumulator(LogContext logContext,
             metrics,
             metricGrpName,
             time,
-            apiVersions,
             transactionManager,
             bufferPool);
     }

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

@@ -16,7 +16,6 @@
  */
 package org.apache.kafka.clients.producer.internals;
 
-import org.apache.kafka.clients.ApiVersions;
 import org.apache.kafka.clients.ClientRequest;
 import org.apache.kafka.clients.ClientResponse;
 import org.apache.kafka.clients.KafkaClient;
@@ -118,9 +117,6 @@ public class Sender implements Runnable {
     /* The max time to wait before retrying a request which has failed */
     private final long retryBackoffMs;
 
-    /* current request API versions supported by the known brokers */
-    private final ApiVersions apiVersions;
-
     /* all the state related to transactions, in particular the producer id, producer epoch, and sequence numbers */
     private final TransactionManager transactionManager;
 
@@ -139,8 +135,7 @@ public Sender(LogContext logContext,
                   Time time,
                   int requestTimeoutMs,
                   long retryBackoffMs,
-                  TransactionManager transactionManager,
-                  ApiVersions apiVersions) {
+                  TransactionManager transactionManager) {
         this.log = logContext.logger(Sender.class);
         this.client = client;
         this.accumulator = accumulator;
@@ -154,7 +149,6 @@ public Sender(LogContext logContext,
         this.sensors = new SenderMetrics(metricsRegistry, metadata, client, time);
         this.requestTimeoutMs = requestTimeoutMs;
         this.retryBackoffMs = retryBackoffMs;
-        this.apiVersions = apiVersions;
         this.transactionManager = transactionManager;
         this.inFlightBatches = new HashMap<>();
     }

Diff for: clients/src/test/java/org/apache/kafka/clients/producer/internals/RecordAccumulatorTest.java

@@ -460,7 +460,7 @@ public void testRetryBackoff() throws Exception {
 
         final RecordAccumulator accum = new RecordAccumulator(logContext, batchSize,
             Compression.NONE, lingerMs, retryBackoffMs, retryBackoffMaxMs,
-            deliveryTimeoutMs, metrics, metricGrpName, time, new ApiVersions(), null,
+            deliveryTimeoutMs, metrics, metricGrpName, time, null,
             new BufferPool(totalSize, batchSize, metrics, time, metricGrpName));
 
         long now = time.milliseconds();
@@ -525,7 +525,7 @@ public void testExponentialRetryBackoff() throws Exception {
 
         final RecordAccumulator accum = new RecordAccumulator(logContext, batchSize,
                 Compression.NONE, lingerMs, retryBackoffMs, retryBackoffMaxMs,
-                deliveryTimeoutMs, metrics, metricGrpName, time, new ApiVersions(), null,
+                deliveryTimeoutMs, metrics, metricGrpName, time, null,
                 new BufferPool(totalSize, batchSize, metrics, time, metricGrpName));
 
         long now = time.milliseconds();
@@ -586,7 +586,7 @@ public void testExponentialRetryBackoffLeaderChange() throws Exception {
 
         final RecordAccumulator accum = new RecordAccumulator(logContext, batchSize,
                 Compression.NONE, lingerMs, retryBackoffMs, retryBackoffMaxMs,
-                deliveryTimeoutMs, metrics, metricGrpName, time, new ApiVersions(), null,
+                deliveryTimeoutMs, metrics, metricGrpName, time, null,
                 new BufferPool(totalSize, batchSize, metrics, time, metricGrpName));
 
         long now = time.milliseconds();
@@ -1266,7 +1266,7 @@ public void testAdaptiveBuiltInPartitioner() throws Exception {
         long totalSize = 1024 * 1024;
         int batchSize = 128;
         RecordAccumulator accum = new RecordAccumulator(logContext, batchSize, Compression.NONE, 0, 0L, 0L,
-                3200, config, metrics, "producer-metrics", time, new ApiVersions(), null,
+                3200, config, metrics, "producer-metrics", time, null,
                 new BufferPool(totalSize, batchSize, metrics, time, "producer-internal-metrics")) {
             @Override
             BuiltInPartitioner createBuiltInPartitioner(LogContext logContext, String topic,
@@ -1399,7 +1399,7 @@ public void testReadyAndDrainWhenABatchIsBeingRetried() throws InterruptedExcept
         String metricGrpName = "producer-metrics";
         final RecordAccumulator accum = new RecordAccumulator(logContext, batchSize,
             Compression.NONE, lingerMs, retryBackoffMs, retryBackoffMaxMs,
-            deliveryTimeoutMs, metrics, metricGrpName, time, new ApiVersions(), null,
+            deliveryTimeoutMs, metrics, metricGrpName, time, null,
             new BufferPool(totalSize, batchSize, metrics, time, metricGrpName));
 
         // Create 1 batch(batchA) to be produced to partition1.
@@ -1661,7 +1661,6 @@ private RecordAccumulator createTestRecordAccumulator(
             metrics,
             metricGrpName,
             time,
-            new ApiVersions(),
             txnManager,
             new BufferPool(totalSize, batchSize, metrics, time, metricGrpName)) {
             @Override