KAFKA-8422; Client should send OffsetForLeaderEpoch only if broker supports latest version

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

@@ -744,6 +744,7 @@ else if (enableAutoCommit)
             Sensor throttleTimeSensor = Fetcher.throttleTimeSensor(metrics, metricsRegistry);
             int heartbeatIntervalMs = config.getInt(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG);
 
+            ApiVersions apiVersions = new ApiVersions();
             NetworkClient netClient = new NetworkClient(
                     new Selector(config.getLong(ConsumerConfig.CONNECTIONS_MAX_IDLE_MS_CONFIG), metrics, time, metricGrpPrefix, channelBuilder, logContext),
                     this.metadata,
@@ -757,7 +758,7 @@ else if (enableAutoCommit)
                     ClientDnsLookup.forConfig(config.getString(ConsumerConfig.CLIENT_DNS_LOOKUP_CONFIG)),
                     time,
                     true,
-                    new ApiVersions(),
+                    apiVersions,
                     throttleTimeSensor,
                     logContext);
             this.client = new ConsumerNetworkClient(
@@ -813,7 +814,8 @@ else if (enableAutoCommit)
                     this.time,
                     this.retryBackoffMs,
                     this.requestTimeoutMs,
-                    isolationLevel);
+                    isolationLevel,
+                    apiVersions);
 
             config.logUnused();
             AppInfoParser.registerAppInfo(JMX_PREFIX, clientId, metrics, time.milliseconds());

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

@@ -16,10 +16,12 @@
  */
 package org.apache.kafka.clients.consumer.internals;
 
+import org.apache.kafka.clients.ApiVersions;
 import org.apache.kafka.clients.ClientResponse;
 import org.apache.kafka.clients.FetchSessionHandler;
 import org.apache.kafka.clients.Metadata;
 import org.apache.kafka.clients.MetadataCache;
+import org.apache.kafka.clients.NodeApiVersions;
 import org.apache.kafka.clients.StaleMetadataException;
 import org.apache.kafka.clients.consumer.ConsumerConfig;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
@@ -49,6 +51,7 @@
 import org.apache.kafka.common.metrics.stats.Meter;
 import org.apache.kafka.common.metrics.stats.Min;
 import org.apache.kafka.common.metrics.stats.Value;
+import org.apache.kafka.common.protocol.ApiKeys;
 import org.apache.kafka.common.protocol.Errors;
 import org.apache.kafka.common.record.BufferSupplier;
 import org.apache.kafka.common.record.ControlRecordType;
@@ -57,13 +60,15 @@
 import org.apache.kafka.common.record.RecordBatch;
 import org.apache.kafka.common.record.Records;
 import org.apache.kafka.common.record.TimestampType;
+import org.apache.kafka.common.requests.ApiVersionsResponse;
 import org.apache.kafka.common.requests.FetchRequest;
 import org.apache.kafka.common.requests.FetchResponse;
 import org.apache.kafka.common.requests.IsolationLevel;
 import org.apache.kafka.common.requests.ListOffsetRequest;
 import org.apache.kafka.common.requests.ListOffsetResponse;
 import org.apache.kafka.common.requests.MetadataRequest;
 import org.apache.kafka.common.requests.MetadataResponse;
+import org.apache.kafka.common.requests.OffsetsForLeaderEpochRequest;
 import org.apache.kafka.common.serialization.Deserializer;
 import org.apache.kafka.common.utils.CloseableIterator;
 import org.apache.kafka.common.utils.LogContext;
@@ -144,6 +149,7 @@ public class Fetcher<K, V> implements Closeable {
     private final AtomicReference<RuntimeException> cachedOffsetForLeaderException = new AtomicReference<>();
     private final OffsetsForLeaderEpochClient offsetsForLeaderEpochClient;
     private final Set<Integer> nodesWithPendingFetchRequests;
+    private final ApiVersions apiVersions;
 
     private PartitionRecords nextInLineRecords = null;
 
@@ -165,7 +171,8 @@ public Fetcher(LogContext logContext,
                    Time time,
                    long retryBackoffMs,
                    long requestTimeoutMs,
-                   IsolationLevel isolationLevel) {
+                   IsolationLevel isolationLevel,
+                   ApiVersions apiVersions) {
         this.log = logContext.logger(Fetcher.class);
         this.logContext = logContext;
         this.time = time;
@@ -186,6 +193,7 @@ public Fetcher(LogContext logContext,
         this.retryBackoffMs = retryBackoffMs;
         this.requestTimeoutMs = requestTimeoutMs;
         this.isolationLevel = isolationLevel;
+        this.apiVersions = apiVersions;
         this.sessionHandlers = new HashMap<>();
         this.offsetsForLeaderEpochClient = new OffsetsForLeaderEpochClient(client, logContext);
         this.nodesWithPendingFetchRequests = new HashSet<>();
@@ -711,6 +719,19 @@ public void onFailure(RuntimeException e) {
         }
     }
 
+    private boolean hasUsableOffsetForLeaderEpochVersion(Node node) {
+        NodeApiVersions nodeApiVersions = apiVersions.get(node.idString());
+
+        if (nodeApiVersions == null)
+            return false;
+
+        ApiVersionsResponse.ApiVersion apiVersion = nodeApiVersions.apiVersion(ApiKeys.OFFSET_FOR_LEADER_EPOCH);
+        if (apiVersion == null)
+            return false;
+
+        return OffsetsForLeaderEpochRequest.supportsTopicPermission(apiVersion.maxVersion);
+    }
+
     /**
      * For each partition which needs validation, make an asynchronous request to get the end-offsets for the partition
      * with the epoch less than or equal to the epoch the partition last saw.
@@ -727,12 +748,22 @@ private void validateOffsetsAsync(Map<TopicPartition, SubscriptionState.FetchPos
                 return;
             }
 
-            subscriptions.setNextAllowedRetry(dataMap.keySet(), time.milliseconds() + requestTimeoutMs);
-
             final Map<TopicPartition, Metadata.LeaderAndEpoch> cachedLeaderAndEpochs = partitionsToValidate.entrySet()
                     .stream()
                     .collect(Collectors.toMap(Map.Entry::getKey, entry -> entry.getValue().currentLeader));
 
+            if (!hasUsableOffsetForLeaderEpochVersion(node)) {
+                log.debug("Skipping validation of fetch offsets for partitions {} since the broker does not " +
+                                "support the required protocol version (introduced in Kafka 2.3)",
+                        cachedLeaderAndEpochs.keySet());
+                for (TopicPartition partition : cachedLeaderAndEpochs.keySet()) {
+                    subscriptions.completeValidation(partition);
+                }
+                return;
+            }
+
+            subscriptions.setNextAllowedRetry(dataMap.keySet(), time.milliseconds() + requestTimeoutMs);
+
             RequestFuture<OffsetsForLeaderEpochClient.OffsetForEpochResult> future = offsetsForLeaderEpochClient.sendAsyncRequest(node, partitionsToValidate);
             future.addListener(new RequestFutureListener<OffsetsForLeaderEpochClient.OffsetForEpochResult>() {
                 @Override
@@ -772,7 +803,7 @@ public void onSuccess(OffsetsForLeaderEpochClient.OffsetForEpochResult offsetsRe
                                 }
                             } else {
                                 // Offset is fine, clear the validation state
-                                subscriptions.validate(respTopicPartition);
+                                subscriptions.completeValidation(respTopicPartition);
                             }
                         }
                     });

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

@@ -19,7 +19,6 @@
 import org.apache.kafka.common.Node;
 import org.apache.kafka.common.TopicPartition;
 import org.apache.kafka.common.errors.TopicAuthorizationException;
-import org.apache.kafka.common.protocol.ApiKeys;
 import org.apache.kafka.common.protocol.Errors;
 import org.apache.kafka.common.requests.AbstractRequest;
 import org.apache.kafka.common.requests.EpochEndOffset;
@@ -53,7 +52,7 @@ protected AbstractRequest.Builder<OffsetsForLeaderEpochRequest> prepareRequest(
             fetchEpoch -> partitionData.put(topicPartition,
                 new OffsetsForLeaderEpochRequest.PartitionData(fetchPosition.currentLeader.epoch, fetchEpoch))));
 
-        return OffsetsForLeaderEpochRequest.Builder.forConsumer(ApiKeys.OFFSET_FOR_LEADER_EPOCH.latestVersion(), partitionData);
+        return OffsetsForLeaderEpochRequest.Builder.forConsumer(partitionData);
     }
 
     @Override

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

@@ -408,7 +408,7 @@ public boolean awaitingValidation(TopicPartition tp) {
         return assignedState(tp).awaitingValidation();
     }
 
-    public void validate(TopicPartition tp) {
+    public void completeValidation(TopicPartition tp) {
         assignedState(tp).validate();
     }
 

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

@@ -84,6 +84,10 @@ public ApiVersion(ApiKeys apiKey) {
             this(apiKey.id, apiKey.oldestVersion(), apiKey.latestVersion());
         }
 
+        public ApiVersion(ApiKeys apiKey, short minVersion, short maxVersion) {
+            this(apiKey.id, minVersion, maxVersion);
+        }
+
         public ApiVersion(short apiKey, short minVersion, short maxVersion) {
             this.apiKey = apiKey;
             this.minVersion = minVersion;

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

@@ -17,6 +17,7 @@
 package org.apache.kafka.common.requests;
 
 import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.errors.UnsupportedVersionException;
 import org.apache.kafka.common.protocol.ApiKeys;
 import org.apache.kafka.common.protocol.Errors;
 import org.apache.kafka.common.protocol.types.Field;
@@ -44,6 +45,11 @@ public class OffsetsForLeaderEpochRequest extends AbstractRequest {
      */
     public static final int CONSUMER_REPLICA_ID = -1;
 
+    /**
+     * Sentinel replica_id which indicates either a debug consumer or a replica which is using
+     * an old version of the protocol.
+     */
+    public static final int DEBUGGING_REPLICA_ID = -2;
 
     private static final Field.ComplexArray TOPICS = new Field.ComplexArray("topics",
             "An array of topics to get epochs for");
@@ -101,23 +107,29 @@ public static class Builder extends AbstractRequest.Builder<OffsetsForLeaderEpoc
         private final Map<TopicPartition, PartitionData> epochsByPartition;
         private final int replicaId;
 
-        Builder(short version, Map<TopicPartition, PartitionData> epochsByPartition, int replicaId) {
-            super(ApiKeys.OFFSET_FOR_LEADER_EPOCH, version);
+        Builder(short oldestAllowedVersion, short latestAllowedVersion, Map<TopicPartition, PartitionData> epochsByPartition, int replicaId) {
+            super(ApiKeys.OFFSET_FOR_LEADER_EPOCH, oldestAllowedVersion, latestAllowedVersion);
             this.epochsByPartition = epochsByPartition;
             this.replicaId = replicaId;
         }
 
-        public static Builder forConsumer(short version, Map<TopicPartition, PartitionData> epochsByPartition) {
-            return new Builder(version, epochsByPartition, CONSUMER_REPLICA_ID);
+        public static Builder forConsumer(Map<TopicPartition, PartitionData> epochsByPartition) {
+            // Old versions of this API require CLUSTER permission which is not typically granted
+            // to clients. Beginning with version 3, the broker requires only TOPIC Describe
+            // permission for the topic of each requested partition. In order to ensure client
+            // compatibility, we only send this request when we can guarantee the relaxed permissions.
+            return new Builder((short) 3, ApiKeys.OFFSET_FOR_LEADER_EPOCH.latestVersion(),
+                    epochsByPartition, CONSUMER_REPLICA_ID);
         }
 
         public static Builder forFollower(short version, Map<TopicPartition, PartitionData> epochsByPartition, int replicaId) {
-            return new Builder(version, epochsByPartition, replicaId);
-
+            return new Builder(version, version, epochsByPartition, replicaId);
         }
 
         @Override
         public OffsetsForLeaderEpochRequest build(short version) {
+            if (version < oldestAllowedVersion() || version > latestAllowedVersion())
+                throw new UnsupportedVersionException("Cannot build " + this + " with version " + version);
             return new OffsetsForLeaderEpochRequest(epochsByPartition, replicaId, version);
         }
 
@@ -143,7 +155,7 @@ public OffsetsForLeaderEpochRequest(Map<TopicPartition, PartitionData> epochsByP
 
     public OffsetsForLeaderEpochRequest(Struct struct, short version) {
         super(ApiKeys.OFFSET_FOR_LEADER_EPOCH, version);
-        replicaId = struct.getOrElse(REPLICA_ID, CONSUMER_REPLICA_ID);
+        replicaId = struct.getOrElse(REPLICA_ID, DEBUGGING_REPLICA_ID);
         epochsByPartition = new HashMap<>();
         for (Object topicAndEpochsObj : struct.get(TOPICS)) {
             Struct topicAndEpochs = (Struct) topicAndEpochsObj;
@@ -222,4 +234,13 @@ public String toString() {
             return bld.toString();
         }
     }
+
+    /**
+     * Check whether a broker allows Topic-level permissions in order to use the
+     * OffsetForLeaderEpoch API. Old versions require Cluster permission.
+     */
+    public static boolean supportsTopicPermission(short latestUsableVersion) {
+        return latestUsableVersion >= 3;
+    }
+
 }

clients/src/main/resources/common/message/OffsetForLeaderEpochRequest.json

@@ -18,12 +18,13 @@
   "type": "request",
   "name": "OffsetForLeaderEpochRequest",
   // Version 1 is the same as version 0.
-  //
   // Version 2 adds the current leader epoch to support fencing.
-  // Version 3 adds ReplicaId
+  // Version 3 adds ReplicaId (the default is -2 which conventionally represents a
+  //    "debug" consumer which is allowed to see offsets beyond the high watermark).
+  //    Followers will use this replicaId when using an older version of the protocol.
   "validVersions": "0-3",
   "fields": [
-    { "name": "ReplicaId", "type": "int32", "versions": "3+",
+    { "name": "ReplicaId", "type": "int32", "versions": "3+", "default": -2, "ignorable": true,
       "about": "The broker ID of the follower, of -1 if this request is from a consumer." },
     { "name": "Topics", "type": "[]OffsetForLeaderTopic", "versions": "0+",
       "about": "Each topic to get offsets for.", "fields": [

clients/src/test/java/org/apache/kafka/clients/ApiVersionsTest.java

@@ -36,7 +36,7 @@ public void testMaxUsableProduceMagic() {
         assertEquals(RecordBatch.CURRENT_MAGIC_VALUE, apiVersions.maxUsableProduceMagic());
 
         apiVersions.update("1", NodeApiVersions.create(Collections.singleton(
-                new ApiVersionsResponse.ApiVersion(ApiKeys.PRODUCE.id, (short) 0, (short) 2))));
+                new ApiVersionsResponse.ApiVersion(ApiKeys.PRODUCE, (short) 0, (short) 2))));
         assertEquals(RecordBatch.MAGIC_VALUE_V1, apiVersions.maxUsableProduceMagic());
 
         apiVersions.remove("1");

clients/src/test/java/org/apache/kafka/clients/NetworkClientTest.java

@@ -272,12 +272,11 @@ public void testConnectionThrottling() {
 
     // Creates expected ApiVersionsResponse from the specified node, where the max protocol version for the specified
     // key is set to the specified version.
-    private ApiVersionsResponse createExpectedApiVersionsResponse(Node node, ApiKeys key,
-        short apiVersionsMaxProtocolVersion) {
+    private ApiVersionsResponse createExpectedApiVersionsResponse(ApiKeys key, short maxVersion) {
         List<ApiVersionsResponse.ApiVersion> versionList = new ArrayList<>();
         for (ApiKeys apiKey : ApiKeys.values()) {
             if (apiKey == key) {
-                versionList.add(new ApiVersionsResponse.ApiVersion(apiKey.id, (short) 0, apiVersionsMaxProtocolVersion));
+                versionList.add(new ApiVersionsResponse.ApiVersion(apiKey, (short) 0, maxVersion));
             } else {
                 versionList.add(new ApiVersionsResponse.ApiVersion(apiKey));
             }
@@ -289,7 +288,7 @@ private ApiVersionsResponse createExpectedApiVersionsResponse(Node node, ApiKeys
     public void testThrottlingNotEnabledForConnectionToOlderBroker() {
         // Instrument the test so that the max protocol version for PRODUCE returned from the node is 5 and thus
         // client-side throttling is not enabled. Also, return a response with a 100ms throttle delay.
-        setExpectedApiVersionsResponse(createExpectedApiVersionsResponse(node, ApiKeys.PRODUCE, (short) 5));
+        setExpectedApiVersionsResponse(createExpectedApiVersionsResponse(ApiKeys.PRODUCE, (short) 5));
         while (!client.ready(node, time.milliseconds()))
             client.poll(1, time.milliseconds());
         selector.clear();

clients/src/test/java/org/apache/kafka/clients/NodeApiVersionsTest.java

@@ -58,7 +58,7 @@ public void testVersionsToString() {
         List<ApiVersion> versionList = new ArrayList<>();
         for (ApiKeys apiKey : ApiKeys.values()) {
             if (apiKey == ApiKeys.DELETE_TOPICS) {
-                versionList.add(new ApiVersion(apiKey.id, (short) 10000, (short) 10001));
+                versionList.add(new ApiVersion(apiKey, (short) 10000, (short) 10001));
             } else {
                 versionList.add(new ApiVersion(apiKey));
             }
@@ -93,7 +93,7 @@ public void testVersionsToString() {
     @Test
     public void testLatestUsableVersion() {
         NodeApiVersions apiVersions = NodeApiVersions.create(Collections.singleton(
-                new ApiVersion(ApiKeys.PRODUCE.id, (short) 1, (short) 3)));
+                new ApiVersion(ApiKeys.PRODUCE, (short) 1, (short) 3)));
         assertEquals(3, apiVersions.latestUsableVersion(ApiKeys.PRODUCE));
         assertEquals(1, apiVersions.latestUsableVersion(ApiKeys.PRODUCE, (short) 0, (short) 1));
         assertEquals(1, apiVersions.latestUsableVersion(ApiKeys.PRODUCE, (short) 1, (short) 1));
@@ -108,14 +108,14 @@ public void testLatestUsableVersion() {
     @Test(expected = UnsupportedVersionException.class)
     public void testLatestUsableVersionOutOfRangeLow() {
         NodeApiVersions apiVersions = NodeApiVersions.create(Collections.singleton(
-                new ApiVersion(ApiKeys.PRODUCE.id, (short) 1, (short) 2)));
+                new ApiVersion(ApiKeys.PRODUCE, (short) 1, (short) 2)));
         apiVersions.latestUsableVersion(ApiKeys.PRODUCE, (short) 3, (short) 4);
     }
 
     @Test(expected = UnsupportedVersionException.class)
     public void testLatestUsableVersionOutOfRangeHigh() {
         NodeApiVersions apiVersions = NodeApiVersions.create(Collections.singleton(
-                new ApiVersion(ApiKeys.PRODUCE.id, (short) 2, (short) 3)));
+                new ApiVersion(ApiKeys.PRODUCE, (short) 2, (short) 3)));
         apiVersions.latestUsableVersion(ApiKeys.PRODUCE, (short) 0, (short) 1);
     }
 
@@ -129,7 +129,7 @@ public void testUsableVersionCalculationNoKnownVersions() {
     @Test(expected = UnsupportedVersionException.class)
     public void testLatestUsableVersionOutOfRange() {
         NodeApiVersions apiVersions = NodeApiVersions.create(Collections.singleton(
-                new ApiVersion(ApiKeys.PRODUCE.id, (short) 300, (short) 300)));
+                new ApiVersion(ApiKeys.PRODUCE, (short) 300, (short) 300)));
         apiVersions.latestUsableVersion(ApiKeys.PRODUCE);
     }
 

clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java

@@ -16,6 +16,7 @@
  */
 package org.apache.kafka.clients.consumer;
 
+import org.apache.kafka.clients.ApiVersions;
 import org.apache.kafka.clients.ClientRequest;
 import org.apache.kafka.clients.KafkaClient;
 import org.apache.kafka.clients.MockClient;
@@ -1924,7 +1925,8 @@ private KafkaConsumer<String, String> newConsumer(Time time,
                 time,
                 retryBackoffMs,
                 requestTimeoutMs,
-                IsolationLevel.READ_UNCOMMITTED);
+                IsolationLevel.READ_UNCOMMITTED,
+                new ApiVersions());
 
         return new KafkaConsumer<>(
                 loggerFactory,