KAFKA-9840: Skip End Offset validation when the leader epoch is not reliable

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

@@ -234,7 +234,9 @@ public synchronized void bootstrap(List<InetSocketAddress> addresses) {
     }
 
     /**
-     * Update metadata assuming the current request version. This is mainly for convenience in testing.
+     * Update metadata assuming the current request version.
+     *
+     * For testing only.
      */
     public synchronized void updateWithCurrentRequestVersion(MetadataResponse response, boolean isPartialUpdate, long nowMs) {
         this.update(this.requestVersion, response, isPartialUpdate, nowMs);
@@ -401,7 +403,7 @@ public synchronized void maybeThrowAnyException() {
      * the producer to abort waiting for metadata if there were fatal exceptions (e.g. authentication failures)
      * in the last metadata update.
      */
-    public synchronized void maybeThrowFatalException() {
+    protected synchronized void maybeThrowFatalException() {
         KafkaException metadataException = this.fatalException;
         if (metadataException != null) {
             fatalException = null;

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

@@ -36,7 +36,8 @@ public class LogTruncationException extends OffsetOutOfRangeException {
     private final Map<TopicPartition, OffsetAndMetadata> divergentOffsets;
 
     public LogTruncationException(Map<TopicPartition, OffsetAndMetadata> divergentOffsets) {
-        super(Utils.transformMap(divergentOffsets, Function.identity(), OffsetAndMetadata::offset));
+        super("Detected log truncation with diverging offsets " + divergentOffsets,
+            Utils.transformMap(divergentOffsets, Function.identity(), OffsetAndMetadata::offset));
         this.divergentOffsets = Collections.unmodifiableMap(divergentOffsets);
     }
 

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

@@ -31,7 +31,12 @@ public class OffsetOutOfRangeException extends InvalidOffsetException {
     private final Map<TopicPartition, Long> offsetOutOfRangePartitions;
 
     public OffsetOutOfRangeException(Map<TopicPartition, Long> offsetOutOfRangePartitions) {
-        super("Offsets out of range with no configured reset policy for partitions: " + offsetOutOfRangePartitions);
+        this("Offsets out of range with no configured reset policy for partitions: " +
+            offsetOutOfRangePartitions, offsetOutOfRangePartitions);
+    }
+
+    public OffsetOutOfRangeException(String message, Map<TopicPartition, Long> offsetOutOfRangePartitions) {
+        super(message);
         this.offsetOutOfRangePartitions = offsetOutOfRangePartitions;
     }
 

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

@@ -85,10 +85,6 @@ protected OffsetForEpochResult handleResponse(
                 case KAFKA_STORAGE_ERROR:
                 case OFFSET_NOT_AVAILABLE:
                 case LEADER_NOT_AVAILABLE:
-                    logger().debug("Attempt to fetch offsets for partition {} failed due to {}, retrying.",
-                            topicPartition, error);
-                    partitionsToRetry.add(topicPartition);
-                    break;
                 case FENCED_LEADER_EPOCH:
                 case UNKNOWN_LEADER_EPOCH:
                     logger().debug("Attempt to fetch offsets for partition {} failed due to {}, retrying.",

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

@@ -786,7 +786,14 @@ private boolean maybeValidatePosition(Metadata.LeaderAndEpoch currentLeaderAndEp
                 return false;
             }
 
-            if (!currentLeaderAndEpoch.leader.isPresent() && !currentLeaderAndEpoch.epoch.isPresent()) {
+            if (!currentLeaderAndEpoch.leader.isPresent()) {
+                return false;
+            }
+
+            // If the leader id is known but the epoch is unknown, we must either have an older API or the epoch has been
+            // discarded. If so, skip the validation and begin fetching.
+            if (!currentLeaderAndEpoch.epoch.isPresent()) {
+                completeValidation();
                 return false;
             }
 
@@ -818,9 +825,7 @@ private void validatePosition(FetchPosition position) {
          */
         private void completeValidation() {
             if (hasPosition()) {
-                transitionState(FetchStates.FETCHING, () -> {
-                    this.nextRetryTimeMs = null;
-                });
+                transitionState(FetchStates.FETCHING, () -> this.nextRetryTimeMs = null);
             }
         }
 
@@ -1011,8 +1016,6 @@ public boolean hasValidPosition() {
      *
      * This includes the offset and epoch from the last record in
      * the batch from a FetchResponse. It also includes the leader epoch at the time the batch was consumed.
-     *
-     * The last fetch epoch is used to
      */
     public static class FetchPosition {
         public final long offset;

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

@@ -40,9 +40,7 @@ public EpochEndOffset(Errors error, int leaderEpoch, long endOffset) {
     }
 
     public EpochEndOffset(int leaderEpoch, long endOffset) {
-        this.error = Errors.NONE;
-        this.leaderEpoch = leaderEpoch;
-        this.endOffset = endOffset;
+        this(Errors.NONE, leaderEpoch, endOffset);
     }
 
     public Errors error() {
@@ -86,4 +84,9 @@ public boolean equals(Object o) {
     public int hashCode() {
         return Objects.hash(error, leaderEpoch, endOffset);
     }
+
+    public boolean hasUndefinedEpochOrOffset() {
+        return this.endOffset == UNDEFINED_EPOCH_OFFSET ||
+            this.leaderEpoch == UNDEFINED_EPOCH;
+    }
 }

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

@@ -371,7 +371,7 @@ public PartitionMetadata withoutLeaderEpoch() {
         @Override
         public String toString() {
             return "PartitionMetadata(" +
-                    ", error=" + error +
+                    "error=" + error +
                     ", partition=" + topicPartition +
                     ", leader=" + leaderId +
                     ", leaderEpoch=" + leaderEpoch +
@@ -429,7 +429,8 @@ private Collection<TopicMetadata> createTopicMetadata(MetadataResponseData data)
 
     public static MetadataResponse prepareResponse(int throttleTimeMs, Collection<Node> brokers, String clusterId,
                                                    int controllerId, List<TopicMetadata> topicMetadataList,
-                                                   int clusterAuthorizedOperations) {
+                                                   int clusterAuthorizedOperations,
+                                                   short responseVersion) {
         MetadataResponseData responseData = new MetadataResponseData();
         responseData.setThrottleTimeMs(throttleTimeMs);
         brokers.forEach(broker ->
@@ -464,22 +465,41 @@ public static MetadataResponse prepareResponse(int throttleTimeMs, Collection<No
             }
             responseData.topics().add(metadataResponseTopic);
         });
-        return new MetadataResponse(responseData);
+        return new MetadataResponse(responseData.toStruct(responseVersion), responseVersion);
     }
 
-    public static MetadataResponse prepareResponse(int throttleTimeMs, Collection<Node> brokers, String clusterId,
-                                                   int controllerId, List<TopicMetadata> topicMetadataList) {
+    public static MetadataResponse prepareResponse(int throttleTimeMs,
+                                                   Collection<Node> brokers,
+                                                   String clusterId,
+                                                   int controllerId,
+                                                   List<TopicMetadata> topicMetadataList,
+                                                   short responseVersion) {
         return prepareResponse(throttleTimeMs, brokers, clusterId, controllerId, topicMetadataList,
-                MetadataResponse.AUTHORIZED_OPERATIONS_OMITTED);
+                MetadataResponse.AUTHORIZED_OPERATIONS_OMITTED, responseVersion);
+    }
+
+    public static MetadataResponse prepareResponse(Collection<Node> brokers,
+                                                   String clusterId,
+                                                   int controllerId,
+                                                   List<TopicMetadata> topicMetadata,
+                                                   short responseVersion) {
+        return prepareResponse(AbstractResponse.DEFAULT_THROTTLE_TIME, brokers, clusterId, controllerId,
+            topicMetadata, responseVersion);
     }
 
-    public static MetadataResponse prepareResponse(Collection<Node> brokers, String clusterId, int controllerId,
+    public static MetadataResponse prepareResponse(Collection<Node> brokers,
+                                                   String clusterId,
+                                                   int controllerId,
                                                    List<TopicMetadata> topicMetadata) {
-        return prepareResponse(AbstractResponse.DEFAULT_THROTTLE_TIME, brokers, clusterId, controllerId, topicMetadata);
+        return prepareResponse(AbstractResponse.DEFAULT_THROTTLE_TIME, brokers, clusterId, controllerId,
+            topicMetadata, ApiKeys.METADATA.latestVersion());
     }
 
-    public static MetadataResponse prepareResponse(int throttleTimeMs, List<MetadataResponseTopic> topicMetadataList,
-                                                   Collection<Node> brokers, String clusterId, int controllerId,
+    public static MetadataResponse prepareResponse(int throttleTimeMs,
+                                                   List<MetadataResponseTopic> topicMetadataList,
+                                                   Collection<Node> brokers,
+                                                   String clusterId,
+                                                   int controllerId,
                                                    int clusterAuthorizedOperations) {
         MetadataResponseData responseData = new MetadataResponseData();
         responseData.setThrottleTimeMs(throttleTimeMs);

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

@@ -47,6 +47,7 @@
 import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
+import java.util.Objects;
 import java.util.Optional;
 import java.util.Set;
 import java.util.concurrent.atomic.AtomicReference;
@@ -141,7 +142,7 @@ public void testTimeToNextUpdate() {
     }
 
     @Test
-    public void testTimeToNextUpdate_RetryBackoff() {
+    public void testTimeToNextUpdateRetryBackoff() {
         long now = 10000;
 
         // lastRefreshMs updated to now.
@@ -200,8 +201,8 @@ public void testIgnoreLeaderEpochInOlderMetadataResponse() {
             assertFalse(response.hasReliableLeaderEpochs());
             metadata.updateWithCurrentRequestVersion(response, false, 100);
             assertTrue(metadata.partitionMetadataIfCurrent(tp).isPresent());
-            MetadataResponse.PartitionMetadata metadata = this.metadata.partitionMetadataIfCurrent(tp).get();
-            assertEquals(Optional.empty(), metadata.leaderEpoch);
+            MetadataResponse.PartitionMetadata responseMetadata = this.metadata.partitionMetadataIfCurrent(tp).get();
+            assertEquals(Optional.empty(), responseMetadata.leaderEpoch);
         }
 
         for (short version = 9; version <= ApiKeys.METADATA.latestVersion(); version++) {
@@ -210,8 +211,8 @@ public void testIgnoreLeaderEpochInOlderMetadataResponse() {
             assertTrue(response.hasReliableLeaderEpochs());
             metadata.updateWithCurrentRequestVersion(response, false, 100);
             assertTrue(metadata.partitionMetadataIfCurrent(tp).isPresent());
-            MetadataResponse.PartitionMetadata info = metadata.partitionMetadataIfCurrent(tp).get();
-            assertEquals(Optional.of(10), info.leaderEpoch);
+            MetadataResponse.PartitionMetadata responseMetadata = metadata.partitionMetadataIfCurrent(tp).get();
+            assertEquals(Optional.of(10), responseMetadata.leaderEpoch);
         }
     }
 
@@ -259,10 +260,10 @@ public void testStaleMetadata() {
         assertEquals(Optional.of(10), metadata.lastSeenLeaderEpoch(tp));
 
         assertTrue(metadata.partitionMetadataIfCurrent(tp).isPresent());
-        MetadataResponse.PartitionMetadata metadata = this.metadata.partitionMetadataIfCurrent(tp).get();
+        MetadataResponse.PartitionMetadata responseMetadata = this.metadata.partitionMetadataIfCurrent(tp).get();
 
-        assertEquals(Arrays.asList(1, 2, 3), metadata.inSyncReplicaIds);
-        assertEquals(Optional.of(10), metadata.leaderEpoch);
+        assertEquals(Arrays.asList(1, 2, 3), responseMetadata.inSyncReplicaIds);
+        assertEquals(Optional.of(10), responseMetadata.leaderEpoch);
     }
 
     @Test
@@ -382,14 +383,16 @@ public void testRejectOldMetadata() {
             MetadataResponse metadataResponse = TestUtils.metadataUpdateWith("dummy", 1, Collections.emptyMap(), partitionCounts, _tp -> 100);
             metadata.updateWithCurrentRequestVersion(metadataResponse, false, 10L);
             assertNotNull(metadata.fetch().partition(tp));
+            assertTrue(metadata.lastSeenLeaderEpoch(tp).isPresent());
             assertEquals(metadata.lastSeenLeaderEpoch(tp).get().longValue(), 100);
         }
 
         // Fake an empty ISR, but with an older epoch, should reject it
         {
             MetadataResponse metadataResponse = TestUtils.metadataUpdateWith("dummy", 1, Collections.emptyMap(), partitionCounts, _tp -> 99,
                 (error, partition, leader, leaderEpoch, replicas, isr, offlineReplicas) ->
-                        new MetadataResponse.PartitionMetadata(error, partition, leader, leaderEpoch, replicas, Collections.emptyList(), offlineReplicas));
+                        new MetadataResponse.PartitionMetadata(error, partition, leader,
+                            leaderEpoch, replicas, Collections.emptyList(), offlineReplicas), ApiKeys.METADATA.latestVersion());
             metadata.updateWithCurrentRequestVersion(metadataResponse, false, 20L);
             assertEquals(metadata.fetch().partition(tp).inSyncReplicas().length, 1);
             assertEquals(metadata.lastSeenLeaderEpoch(tp).get().longValue(), 100);
@@ -399,7 +402,8 @@ public void testRejectOldMetadata() {
         {
             MetadataResponse metadataResponse = TestUtils.metadataUpdateWith("dummy", 1, Collections.emptyMap(), partitionCounts, _tp -> 100,
                 (error, partition, leader, leaderEpoch, replicas, isr, offlineReplicas) ->
-                        new MetadataResponse.PartitionMetadata(error, partition, leader, leaderEpoch, replicas, Collections.emptyList(), offlineReplicas));
+                        new MetadataResponse.PartitionMetadata(error, partition, leader,
+                            leaderEpoch, replicas, Collections.emptyList(), offlineReplicas), ApiKeys.METADATA.latestVersion());
             metadata.updateWithCurrentRequestVersion(metadataResponse, false, 20L);
             assertEquals(metadata.fetch().partition(tp).inSyncReplicas().length, 0);
             assertEquals(metadata.lastSeenLeaderEpoch(tp).get().longValue(), 100);
@@ -436,29 +440,30 @@ public void testOutOfBandEpochUpdate() {
         MetadataResponse metadataResponse = TestUtils.metadataUpdateWith("dummy", 1, Collections.emptyMap(), partitionCounts, _tp -> 100);
         metadata.updateWithCurrentRequestVersion(metadataResponse, false, 10L);
         assertNotNull(metadata.fetch().partition(tp));
+        assertTrue(metadata.lastSeenLeaderEpoch(tp).isPresent());
         assertEquals(metadata.lastSeenLeaderEpoch(tp).get().longValue(), 100);
 
         // Simulate a leader epoch from another response, like a fetch response or list offsets
         assertTrue(metadata.updateLastSeenEpochIfNewer(tp, 101));
 
         // Cache of partition stays, but current partition info is not available since it's stale
         assertNotNull(metadata.fetch().partition(tp));
-        assertEquals(metadata.fetch().partitionCountForTopic("topic-1").longValue(), 5);
+        assertEquals(Objects.requireNonNull(metadata.fetch().partitionCountForTopic("topic-1")).longValue(), 5);
         assertFalse(metadata.partitionMetadataIfCurrent(tp).isPresent());
         assertEquals(metadata.lastSeenLeaderEpoch(tp).get().longValue(), 101);
 
         // Metadata with older epoch is rejected, metadata state is unchanged
         metadata.updateWithCurrentRequestVersion(metadataResponse, false, 20L);
         assertNotNull(metadata.fetch().partition(tp));
-        assertEquals(metadata.fetch().partitionCountForTopic("topic-1").longValue(), 5);
+        assertEquals(Objects.requireNonNull(metadata.fetch().partitionCountForTopic("topic-1")).longValue(), 5);
         assertFalse(metadata.partitionMetadataIfCurrent(tp).isPresent());
         assertEquals(metadata.lastSeenLeaderEpoch(tp).get().longValue(), 101);
 
         // Metadata with equal or newer epoch is accepted
         metadataResponse = TestUtils.metadataUpdateWith("dummy", 1, Collections.emptyMap(), partitionCounts, _tp -> 101);
         metadata.updateWithCurrentRequestVersion(metadataResponse, false, 30L);
         assertNotNull(metadata.fetch().partition(tp));
-        assertEquals(metadata.fetch().partitionCountForTopic("topic-1").longValue(), 5);
+        assertEquals(Objects.requireNonNull(metadata.fetch().partitionCountForTopic("topic-1")).longValue(), 5);
         assertTrue(metadata.partitionMetadataIfCurrent(tp).isPresent());
         assertEquals(metadata.lastSeenLeaderEpoch(tp).get().longValue(), 101);
     }
@@ -706,7 +711,7 @@ public void testNodeIfOffline() {
             (error, partition, leader, leaderEpoch, replicas, isr, offlineReplicas) ->
                 new MetadataResponse.PartitionMetadata(error, partition, Optional.of(node0.id()), leaderEpoch,
                     Collections.singletonList(node0.id()), Collections.emptyList(),
-                        Collections.singletonList(node1.id())));
+                        Collections.singletonList(node1.id())), ApiKeys.METADATA.latestVersion());
         metadata.updateWithCurrentRequestVersion(emptyMetadataResponse(), false, 0L);
         metadata.updateWithCurrentRequestVersion(metadataResponse, false, 10L);
 

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

@@ -611,7 +611,7 @@ public void testHeartbeatIllegalGenerationResponseWithOldGeneration() throws Int
 
         final AbstractCoordinator.Generation currGen = coordinator.generation();
 
-        // let the heartbeat request to send out a request
+        // let the heartbeat thread send out a request
         mockTime.sleep(HEARTBEAT_INTERVAL_MS);
 
         TestUtils.waitForCondition(() -> !mockClient.requests().isEmpty(), 2000,