KAFKA-9724 Newer clients not always sending fetch request to older brokers

clients/src/main/java/org/apache/kafka/clients/Metadata.java

@@ -163,7 +163,8 @@ public synchronized boolean updateLastSeenEpochIfNewer(TopicPartition topicParti
         if (leaderEpoch < 0)
             throw new IllegalArgumentException("Invalid leader epoch " + leaderEpoch + " (must be non-negative)");
 
-        boolean updated = updateLastSeenEpoch(topicPartition, leaderEpoch, oldEpoch -> leaderEpoch > oldEpoch);
+        // We don't want to overwrite a null epoch with an epoch from something other than a metadata update
+        boolean updated = updateLastSeenEpoch(topicPartition, leaderEpoch, oldEpoch -> leaderEpoch > oldEpoch, false);
         this.needFullUpdate = this.needFullUpdate || updated;
         return updated;
     }
@@ -178,11 +179,13 @@ public Optional<Integer> lastSeenLeaderEpoch(TopicPartition topicPartition) {
      * @param topicPartition       topic+partition to update the epoch for
      * @param epoch                the new epoch
      * @param epochTest            a predicate to determine if the old epoch should be replaced
+     * @param overwriteNullEpoch   if we should allow overwriting a missing epoch
      * @return true if the epoch was updated, false otherwise
      */
     private synchronized boolean updateLastSeenEpoch(TopicPartition topicPartition,
                                                      int epoch,
-                                                     Predicate<Integer> epochTest) {
+                                                     Predicate<Integer> epochTest,
+                                                     boolean overwriteNullEpoch) {
         Integer oldEpoch = lastSeenLeaderEpochs.get(topicPartition);
         log.trace("Determining if we should replace existing epoch {} with new epoch {}", oldEpoch, epoch);
         if (oldEpoch == null || epochTest.test(oldEpoch)) {
@@ -373,7 +376,7 @@ private Optional<MetadataResponse.PartitionMetadata> updateLatestMetadata(
         if (hasReliableLeaderEpoch && partitionMetadata.leaderEpoch.isPresent()) {
             int newEpoch = partitionMetadata.leaderEpoch.get();
             // If the received leader epoch is at least the same as the previous one, update the metadata
-            if (updateLastSeenEpoch(tp, newEpoch, oldEpoch -> newEpoch >= oldEpoch)) {
+            if (updateLastSeenEpoch(tp, newEpoch, oldEpoch -> newEpoch >= oldEpoch, true)) {
                 return Optional.of(partitionMetadata);
             } else {
                 // Otherwise ignore the new metadata and use the previously cached info

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

@@ -1096,9 +1096,26 @@ Node selectReadReplica(TopicPartition partition, Node leaderReplica, long curren
     private Map<Node, FetchSessionHandler.FetchRequestData> prepareFetchRequests() {
         Map<Node, FetchSessionHandler.Builder> fetchable = new LinkedHashMap<>();
 
-        // Ensure the position has an up-to-date leader
-        subscriptions.assignedPartitions().forEach(
-            tp -> subscriptions.maybeValidatePositionForCurrentLeader(tp, metadata.currentLeader(tp)));
+        // Ensure the position has an up-to-date leader, if the leader is set and it's ApiVersion is new enough
+        subscriptions.assignedPartitions().forEach(tp -> {
+            Metadata.LeaderAndEpoch leaderAndEpoch = metadata.currentLeader(tp);
+            final boolean offsetForEpochAvailable;
+            if (leaderAndEpoch.leader.isPresent()) {
+                NodeApiVersions nodeApiVersions = apiVersions.get(leaderAndEpoch.leader.get().idString());
+                if (nodeApiVersions == null) {
+                    client.tryConnect(leaderAndEpoch.leader.get());
+                    return;
+                }
+                offsetForEpochAvailable = hasUsableOffsetForLeaderEpochVersion(nodeApiVersions);
+            } else {
+                offsetForEpochAvailable = false;
+            }
+            if (offsetForEpochAvailable) {
+                subscriptions.maybeValidatePositionForCurrentLeader(tp, metadata.currentLeader(tp));
+            } else {
+                subscriptions.completeValidation(tp);
+            }
+        });
 
         long currentTimeMs = time.milliseconds();
 

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

@@ -3617,18 +3617,35 @@ public void testOffsetValidationSkippedForOldBroker() {
         apiVersions.update(node.idString(), NodeApiVersions.create(
             ApiKeys.OFFSET_FOR_LEADER_EPOCH.id, (short) 0, (short) 2));
 
-        // Seek with a position and leader+epoch
-        Metadata.LeaderAndEpoch leaderAndEpoch = new Metadata.LeaderAndEpoch(
-                metadata.currentLeader(tp0).leader, Optional.of(epochOne));
-        subscriptions.seekUnvalidated(tp0, new SubscriptionState.FetchPosition(0, Optional.of(epochOne), leaderAndEpoch));
+        {
+            // Seek with a position and leader+epoch
+            Metadata.LeaderAndEpoch leaderAndEpoch = new Metadata.LeaderAndEpoch(
+                    metadata.currentLeader(tp0).leader, Optional.of(epochOne));
+            subscriptions.seekUnvalidated(tp0, new SubscriptionState.FetchPosition(0, Optional.of(epochOne), leaderAndEpoch));
+
+            // Update metadata to epoch=2, enter validation
+            metadata.updateWithCurrentRequestVersion(TestUtils.metadataUpdateWith("dummy", 1,
+                    Collections.emptyMap(), partitionCounts, tp -> epochTwo), false, 0L);
+            fetcher.validateOffsetsIfNeeded();
+
+            // Offset validation is skipped
+            assertFalse(subscriptions.awaitingValidation(tp0));
+        }
 
-        // Update metadata to epoch=2, enter validation
-        metadata.updateWithCurrentRequestVersion(TestUtils.metadataUpdateWith("dummy", 1,
-                Collections.emptyMap(), partitionCounts, tp -> epochTwo), false, 0L);
-        fetcher.validateOffsetsIfNeeded();
+        {
+            // Seek with a position and leader+epoch
+            Metadata.LeaderAndEpoch leaderAndEpoch = new Metadata.LeaderAndEpoch(
+                    metadata.currentLeader(tp0).leader, Optional.of(epochOne));
+            subscriptions.seekUnvalidated(tp0, new SubscriptionState.FetchPosition(0, Optional.of(epochOne), leaderAndEpoch));
 
-        // Offset validation is skipped
-        assertFalse(subscriptions.awaitingValidation(tp0));
+            // Update metadata to epoch=2, enter validation
+            metadata.updateWithCurrentRequestVersion(TestUtils.metadataUpdateWith("dummy", 1,
+                    Collections.emptyMap(), partitionCounts, tp -> epochTwo), false, 0L);
+
+            // Subscription should not stay in AWAITING_VALIDATION in prepareFetchRequest
+            assertEquals(1, fetcher.sendFetches());
+            assertFalse(subscriptions.awaitingValidation(tp0));
+        }
     }
 
     @Test