MINOR: Cleanup 0.10.x legacy references in ClusterResourceListener and TopicConfig (clients module) (apache#19388)

This PR is a minor follow-up to [PR
apache#19320](apache#19320), which cleaned up
0.10.x legacy information from the clients module.

It addresses remaining reviewer suggestions that were not included in
the original PR:

- `ClusterResourceListener`: Removed "Note the minimum supported broker
version is 2.1." per review suggestion to avoid repeating
version-specific details across multiple classes.
- `TopicConfig`: Simplified `MAX_MESSAGE_BYTES_DOC` by removing obsolete
notes about behavior in versions prior to 0.10.2.

These changes help reduce outdated version information in client
documentation and improve clarity.

Reviewers: PoAn Yang <[email protected]>, Chia-Ping Tsai
<[email protected]>

Author: LoganZhuZzz

clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerConfig.java

@@ -371,8 +371,7 @@ public class ConsumerConfig extends AbstractConfig {
     public static final String ALLOW_AUTO_CREATE_TOPICS_CONFIG = "allow.auto.create.topics";
     private static final String ALLOW_AUTO_CREATE_TOPICS_DOC = "Allow automatic topic creation on the broker when" +
             " subscribing to or assigning a topic. A topic being subscribed to will be automatically created only if the" +
-            " broker allows for it using `auto.create.topics.enable` broker configuration. This configuration must" +
-            " be set to `true` when using brokers older than 0.11.0";
+            " broker allows for it using `auto.create.topics.enable` broker configuration.";
     public static final boolean DEFAULT_ALLOW_AUTO_CREATE_TOPICS = true;
 
     /**

clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractFetch.java

@@ -218,8 +218,7 @@ protected void handleFetchSuccess(final Node fetchTarget,
                         partition,
                         partitionData,
                         metricAggregator,
-                        fetchOffset,
-                        requestVersion);
+                        fetchOffset);
                 fetchBuffer.add(completedFetch);
             }
 

clients/src/main/java/org/apache/kafka/clients/consumer/internals/CompletedFetch.java

@@ -60,7 +60,6 @@ public class CompletedFetch {
 
     final TopicPartition partition;
     final FetchResponseData.PartitionData partitionData;
-    final short requestVersion;
 
     private final Logger log;
     private final SubscriptionState subscriptions;
@@ -88,8 +87,7 @@ public class CompletedFetch {
                    TopicPartition partition,
                    FetchResponseData.PartitionData partitionData,
                    FetchMetricsAggregator metricAggregator,
-                   Long fetchOffset,
-                   short requestVersion) {
+                   Long fetchOffset) {
         this.log = log;
         this.subscriptions = subscriptions;
         this.decompressionBufferSupplier = decompressionBufferSupplier;
@@ -98,7 +96,6 @@ public class CompletedFetch {
         this.metricAggregator = metricAggregator;
         this.batches = FetchResponse.recordsOrFail(partitionData).batches().iterator();
         this.nextFetchOffset = fetchOffset;
-        this.requestVersion = requestVersion;
         this.lastEpoch = Optional.empty();
         this.abortedProducerIds = new HashSet<>();
         this.abortedTransactions = abortedTransactions(partitionData);

clients/src/main/java/org/apache/kafka/clients/consumer/internals/FetchCollector.java

@@ -16,13 +16,11 @@
  */
 package org.apache.kafka.clients.consumer.internals;
 
-import org.apache.kafka.clients.consumer.ConsumerConfig;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.apache.kafka.clients.consumer.OffsetAndMetadata;
 import org.apache.kafka.clients.consumer.OffsetOutOfRangeException;
 import org.apache.kafka.common.KafkaException;
 import org.apache.kafka.common.TopicPartition;
-import org.apache.kafka.common.errors.RecordTooLargeException;
 import org.apache.kafka.common.errors.TopicAuthorizationException;
 import org.apache.kafka.common.message.FetchResponseData;
 import org.apache.kafka.common.protocol.Errors;
@@ -37,7 +35,6 @@
 import java.util.Collections;
 import java.util.Iterator;
 import java.util.List;
-import java.util.Map;
 import java.util.Optional;
 import java.util.Queue;
 
@@ -263,21 +260,10 @@ private CompletedFetch handleInitializeSuccess(final CompletedFetch completedFet
         Iterator<? extends RecordBatch> batches = FetchResponse.recordsOrFail(partition).batches().iterator();
 
         if (!batches.hasNext() && FetchResponse.recordsSize(partition) > 0) {
-            if (completedFetch.requestVersion < 3) {
-                // Implement the pre KIP-74 behavior of throwing a RecordTooLargeException.
-                Map<TopicPartition, Long> recordTooLargePartitions = Collections.singletonMap(tp, fetchOffset);
-                throw new RecordTooLargeException("There are some messages at [Partition=Offset]: " +
-                        recordTooLargePartitions + " whose size is larger than the fetch size " + fetchConfig.fetchSize +
-                        " and hence cannot be returned. Please considering upgrading your broker to 0.10.1.0 or " +
-                        "newer to avoid this issue. Alternately, increase the fetch size on the client (using " +
-                        ConsumerConfig.MAX_PARTITION_FETCH_BYTES_CONFIG + ")",
-                        recordTooLargePartitions);
-            } else {
-                // This should not happen with brokers that support FetchRequest/Response V3 or higher (i.e. KIP-74)
-                throw new KafkaException("Failed to make progress reading messages at " + tp + "=" +
-                        fetchOffset + ". Received a non-empty fetch response from the server, but no " +
-                        "complete records were found.");
-            }
+            // This should not happen with brokers that support FetchRequest/Response V4 or higher (i.e. KIP-74)
+            throw new KafkaException("Failed to make progress reading messages at " + tp + "=" +
+                    fetchOffset + ". Received a non-empty fetch response from the server, but no " +
+                    "complete records were found.");
         }
 
         if (!updatePartitionState(partition, tp)) {

clients/src/main/java/org/apache/kafka/common/ClusterResourceListener.java

@@ -24,7 +24,6 @@
  * <p>
  * <h4>Clients</h4>
  * There will be one invocation of {@link ClusterResourceListener#onUpdate(ClusterResource)} after each metadata response.
- * Note the minimum supported broker version is 2.1.
  * <p>
  * {@link org.apache.kafka.clients.producer.ProducerInterceptor} : The {@link ClusterResourceListener#onUpdate(ClusterResource)} method will be invoked after {@link org.apache.kafka.clients.producer.ProducerInterceptor#onSend(org.apache.kafka.clients.producer.ProducerRecord)}
  * but before {@link org.apache.kafka.clients.producer.ProducerInterceptor#onAcknowledgement(org.apache.kafka.clients.producer.RecordMetadata, Exception)} .

clients/src/main/java/org/apache/kafka/common/config/TopicConfig.java

@@ -108,10 +108,7 @@ public class TopicConfig {
 
     public static final String MAX_MESSAGE_BYTES_CONFIG = "max.message.bytes";
     public static final String MAX_MESSAGE_BYTES_DOC =
-        "The largest record batch size allowed by Kafka (after compression if compression is enabled). " +
-        "In the latest message format version, records are always grouped into batches for efficiency. " +
-        "In previous message format versions, uncompressed records are not grouped into batches and this " +
-        "limit only applies to a single record in that case.";
+        "The largest record batch size allowed by Kafka (after compression if compression is enabled).";
 
     public static final String INDEX_INTERVAL_BYTES_CONFIG = "index.interval.bytes";
     public static final String INDEX_INTERVAL_BYTES_DOC = "This setting controls how frequently " +

clients/src/test/java/org/apache/kafka/clients/consumer/internals/CompletedFetchTest.java

@@ -26,7 +26,6 @@
 import org.apache.kafka.common.header.internals.RecordHeaders;
 import org.apache.kafka.common.message.FetchResponseData;
 import org.apache.kafka.common.metrics.Metrics;
-import org.apache.kafka.common.protocol.ApiKeys;
 import org.apache.kafka.common.record.ControlRecordType;
 import org.apache.kafka.common.record.EndTransactionMarker;
 import org.apache.kafka.common.record.MemoryRecords;
@@ -227,8 +226,7 @@ private CompletedFetch newCompletedFetch(long fetchOffset,
                 TP,
                 partitionData,
                 metricAggregator,
-                fetchOffset,
-                ApiKeys.FETCH.latestVersion());
+                fetchOffset);
     }
 
     private static Deserializers<UUID, UUID> newUuidDeserializers() {

clients/src/test/java/org/apache/kafka/clients/consumer/internals/FetchBufferTest.java

@@ -20,7 +20,6 @@
 import org.apache.kafka.common.TopicPartition;
 import org.apache.kafka.common.message.FetchResponseData;
 import org.apache.kafka.common.metrics.Metrics;
-import org.apache.kafka.common.protocol.ApiKeys;
 import org.apache.kafka.common.serialization.StringSerializer;
 import org.apache.kafka.common.utils.BufferSupplier;
 import org.apache.kafka.common.utils.LogContext;
@@ -198,8 +197,7 @@ private CompletedFetch completedFetch(TopicPartition tp) {
                 tp,
                 partitionData,
                 metricsAggregator,
-                0L,
-                ApiKeys.FETCH.latestVersion());
+                0L);
     }
 
     /**

clients/src/test/java/org/apache/kafka/clients/consumer/internals/FetchCollectorTest.java

@@ -26,7 +26,6 @@
 import org.apache.kafka.common.internals.ClusterResourceListeners;
 import org.apache.kafka.common.message.FetchResponseData;
 import org.apache.kafka.common.metrics.Metrics;
-import org.apache.kafka.common.protocol.ApiKeys;
 import org.apache.kafka.common.protocol.Errors;
 import org.apache.kafka.common.record.ControlRecordType;
 import org.apache.kafka.common.record.EndTransactionMarker;
@@ -921,8 +920,7 @@ private CompletedFetch build() {
                     topicPartition,
                     partitionData,
                     metricsAggregator,
-                    fetchOffset,
-                    ApiKeys.FETCH.latestVersion());
+                    fetchOffset);
         }
     }