KAFKA-15615: Improve handling of fetching during metadata updates

- Add test cases about REPLICA_NOT_AVAILABLE error
- Fix some comments

Author: appchemist

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

@@ -3213,7 +3213,7 @@ public void testWhenFetchResponseReturnsALeaderShipChangeErrorButNoNewLeaderInfo
         subscriptions.seek(tp0, 0);
         subscriptions.seek(tp1, 0);
 
-        // Setup preferred read replica to node=1 by doing a fetch for both partitions.
+        // Setup preferred read replica to node=0 by doing a fetch for both partitions.
         assertEquals(2, sendFetches());
         assertFalse(fetcher.hasCompletedFetches());
         client.prepareResponseFrom(fullFetchResponse(tidp0, this.records, Errors.NONE, 100L,
@@ -3310,7 +3310,7 @@ public void testWhenFetchResponseReturnsALeaderShipChangeErrorButNoNewLeaderInfo
         assertEquals(Optional.of(nodeId1.id()), subscriptions.preferredReadReplica(tp0, time.milliseconds()));
         assertEquals(Optional.of(nodeId1.id()), subscriptions.preferredReadReplica(tp1, time.milliseconds()));
 
-        // Validate subscription is valid & fetch-able, and points to the new leader.
+        // Validate subscription is valid & fetch-able, and points to the new leader info.
         assertTrue(subscriptions.isFetchable(tp0));
         currentLeader = subscriptions.position(tp0).currentLeader;
         assertEquals(tp0Leader.id(), currentLeader.leader.get().id());
@@ -3347,7 +3347,7 @@ public void testWhenFetchResponseReturnsALeaderShipChangeErrorAndNewLeaderInform
         subscriptions.seek(tp0, 0);
         subscriptions.seek(tp1, 0);
 
-        // Setup preferred read replica to node=1 by doing a fetch for both partitions.
+        // Setup preferred read replica to node=0 by doing a fetch for both partitions.
         assertEquals(2, sendFetches());
         assertFalse(fetcher.hasCompletedFetches());
         client.prepareResponseFrom(fullFetchResponse(tidp0, this.records, Errors.NONE, 100L,
@@ -3428,9 +3428,9 @@ public void testWhenFetchResponseReturnsALeaderShipChangeErrorAndNewLeaderInform
     @ParameterizedTest
     @EnumSource(value = Errors.class, names = {"KAFKA_STORAGE_ERROR", "OFFSET_NOT_AVAILABLE",
             "UNKNOWN_TOPIC_OR_PARTITION", "UNKNOWN_TOPIC_ID", "INCONSISTENT_TOPIC_ID"})
-    public void testWhenFetchResponseReturnsAErrorCausingRequestingMetadata(Errors error) {
+    public void testWhenFetchResponseReturnsAErrorCausingAwaitUpdate(Errors error) {
         // The test runs with 2 partitions where 1 partition is fetched without errors, and
-        // 2nd partition faces errors due to leadership changes.
+        // 2nd partition faces errors.
         buildFetcher(new MetricConfig(), OffsetResetStrategy.EARLIEST, new BytesDeserializer(),
                 new BytesDeserializer(),
                 Integer.MAX_VALUE, IsolationLevel.READ_UNCOMMITTED,
@@ -3449,7 +3449,7 @@ public void testWhenFetchResponseReturnsAErrorCausingRequestingMetadata(Errors e
         subscriptions.seek(tp0, 0);
         subscriptions.seek(tp1, 0);
 
-        // Setup preferred read replica to node=1 by doing a fetch for both partitions.
+        // Setup preferred read replica to node=0 by doing a fetch for both partitions.
         assertEquals(2, sendFetches());
         assertFalse(fetcher.hasCompletedFetches());
         client.prepareResponseFrom(fullFetchResponse(tidp0, this.records, Errors.NONE, 100L,
@@ -3473,8 +3473,7 @@ public void testWhenFetchResponseReturnsAErrorCausingRequestingMetadata(Errors e
         // Verify that metadata-update isn't requested as metadata is considered upto-date.
         assertFalse(metadata.updateRequested());
 
-        // TEST that next fetch returns an error(due to leadership change) but new leader info is not returned
-        // in the FetchResponse. This is the behaviour prior to KIP-951, should keep on working.
+        // TEST that next fetch returns an error.
         LinkedHashMap<TopicIdPartition, FetchResponseData.PartitionData> partitions = new LinkedHashMap<>();
         partitions.put(tidp0,
                 new FetchResponseData.PartitionData()
@@ -3494,10 +3493,10 @@ public void testWhenFetchResponseReturnsAErrorCausingRequestingMetadata(Errors e
         assertFalse(partitionRecords.containsKey(tp0));
         assertTrue(partitionRecords.containsKey(tp1));
 
-        // Validate metadata is unchanged, as FetchResponse didn't have new leader information.
+        // Validate metadata is unchanged.
         assertEquals(startingClusterMetadata, metadata.fetch());
 
-        // Validate metadata-update is requested due to the leadership-error on tp0.
+        // Validate metadata-update is requested due to the error on tp0.
         assertTrue(metadata.updateRequested());
 
         // Validate preferred-read-replica is cleared for tp0 due to the error.
@@ -3546,7 +3545,7 @@ public void testWhenFetchResponseReturnsAErrorCausingRequestingMetadata(Errors e
         assertEquals(Optional.of(nodeId1.id()), subscriptions.preferredReadReplica(tp0, time.milliseconds()));
         assertEquals(Optional.of(nodeId1.id()), subscriptions.preferredReadReplica(tp1, time.milliseconds()));
 
-        // Validate subscription is valid & fetch-able, and points to the new leader.
+        // Validate subscription is valid & fetch-able, and points to the new leader info.
         assertTrue(subscriptions.isFetchable(tp0));
         currentLeader = subscriptions.position(tp0).currentLeader;
         assertEquals(tp0Leader.id(), currentLeader.leader.get().id());
@@ -3556,6 +3555,97 @@ public void testWhenFetchResponseReturnsAErrorCausingRequestingMetadata(Errors e
         assertTrue(subscriptions.isFetchable(tp1));
     }
 
+    @Test
+    public void testWhenFetchResponseReturnsAReplicaNotAvailableError() {
+        Errors error = Errors.REPLICA_NOT_AVAILABLE;
+
+        // The test runs with 2 partitions where 1 partition is fetched without errors, and
+        // 2nd partition faces errors due to replica not available.
+        buildFetcher(new MetricConfig(), OffsetResetStrategy.EARLIEST, new BytesDeserializer(),
+                new BytesDeserializer(),
+                Integer.MAX_VALUE, IsolationLevel.READ_UNCOMMITTED,
+                Duration.ofMinutes(5).toMillis());
+
+        // Setup so that tp0 & tp1 are subscribed and will be fetched from.
+        // Also, setup client's metadata for tp0 & tp1.
+        subscriptions.assignFromUser(new HashSet<>(Arrays.asList(tp0, tp1)));
+        client.updateMetadata(
+                RequestTestUtils.metadataUpdateWithIds(2, singletonMap(topicName, 4),
+                        tp -> validLeaderEpoch, topicIds, false));
+        Node tp0Leader = metadata.fetch().leaderFor(tp0);
+        Node tp1Leader = metadata.fetch().leaderFor(tp1);
+        Node nodeId0 = metadata.fetch().nodeById(0);
+        Cluster startingClusterMetadata = metadata.fetch();
+        subscriptions.seek(tp0, 0);
+        subscriptions.seek(tp1, 0);
+
+        // Setup preferred read replica to node=0 by doing a fetch for both partitions.
+        assertEquals(2, sendFetches());
+        assertFalse(fetcher.hasCompletedFetches());
+        client.prepareResponseFrom(fullFetchResponse(tidp0, this.records, Errors.NONE, 100L,
+                FetchResponse.INVALID_LAST_STABLE_OFFSET, 0, Optional.of(nodeId0.id())), tp0Leader);
+        client.prepareResponseFrom(fullFetchResponse(tidp1, this.records, Errors.NONE, 100L,
+                FetchResponse.INVALID_LAST_STABLE_OFFSET, 0, Optional.of(nodeId0.id())), tp1Leader);
+        networkClientDelegate.poll(time.timer(0));
+        assertTrue(fetcher.hasCompletedFetches());
+        Map<TopicPartition, List<ConsumerRecord<byte[], byte[]>>> partitionRecords = fetchRecords();
+        assertTrue(partitionRecords.containsKey(tp0));
+        assertTrue(partitionRecords.containsKey(tp1));
+        // Validate setup of preferred read replica for tp0 & tp1 is done correctly.
+        Node selected = fetcher.selectReadReplica(tp0, Node.noNode(), time.milliseconds());
+        assertEquals(nodeId0.id(), selected.id());
+        selected = fetcher.selectReadReplica(tp1, Node.noNode(), time.milliseconds());
+        assertEquals(nodeId0.id(), selected.id());
+
+        // Send next fetch request.
+        assertEquals(1, sendFetches());
+        assertFalse(fetcher.hasCompletedFetches());
+        // Verify that metadata-update isn't requested as metadata is considered upto-date.
+        assertFalse(metadata.updateRequested());
+
+        // TEST that next fetch returns an error.
+        LinkedHashMap<TopicIdPartition, FetchResponseData.PartitionData> partitions = new LinkedHashMap<>();
+        partitions.put(tidp0,
+                new FetchResponseData.PartitionData()
+                        .setPartitionIndex(tidp0.topicPartition().partition())
+                        .setErrorCode(error.code()));
+        partitions.put(tidp1,
+                new FetchResponseData.PartitionData()
+                        .setPartitionIndex(tidp1.topicPartition().partition())
+                        .setErrorCode(Errors.NONE.code())
+                        .setHighWatermark(100L)
+                        .setLastStableOffset(FetchResponse.INVALID_LAST_STABLE_OFFSET)
+                        .setLogStartOffset(0)
+                        .setRecords(nextRecords));
+        client.prepareResponseFrom(FetchResponse.of(Errors.NONE, 0, INVALID_SESSION_ID, partitions), nodeId0);
+        networkClientDelegate.poll(time.timer(0));
+        partitionRecords = fetchRecords();
+        assertFalse(partitionRecords.containsKey(tp0));
+        assertTrue(partitionRecords.containsKey(tp1));
+
+        // Validate metadata is unchanged.
+        assertEquals(startingClusterMetadata, metadata.fetch());
+
+        // Validate metadata-update is requested due to the replica not available error on tp0.
+        assertTrue(metadata.updateRequested());
+
+        // Validate preferred-read-replica is cleared for tp0 due to the error.
+        assertEquals(Optional.empty(),
+                subscriptions.preferredReadReplica(tp0, time.milliseconds()));
+        // Validate preferred-read-replica is still set for tp1 as previous fetch for it was ok.
+        assertEquals(Optional.of(nodeId0.id()),
+                subscriptions.preferredReadReplica(tp1, time.milliseconds()));
+
+        // Validate subscription is still valid & fetch-able for tp0 & tp1.
+        // And tp0 points to original leader.
+        assertTrue(subscriptions.isFetchable(tp0));
+        Metadata.LeaderAndEpoch currentLeader = subscriptions.position(tp0).currentLeader;
+        assertEquals(tp0Leader.id(), currentLeader.leader.get().id());
+        assertEquals(validLeaderEpoch, currentLeader.epoch.get());
+
+        assertTrue(subscriptions.isFetchable(tp1));
+    }
+
     private OffsetsForLeaderEpochResponse prepareOffsetsForLeaderEpochResponse(
             TopicPartition topicPartition,
             Errors error,