@ParameterizedTest
    @EnumSource(value = Errors.class, names = {"FENCED_LEADER_EPOCH", "NOT_LEADER_OR_FOLLOWER"})
    public void testWhenFetchResponseReturnsALeaderShipChangeErrorAndNewLeaderInformation(Errors error) {
        // The test runs with 2 partitions where 1 partition is fetched without errors, and
        // 2nd partition faces errors due to leadership changes.
        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=1 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());
        // Validate metadata-update isn't requested as no errors seen yet.
        assertFalse(metadata.updateRequested());

        // Test that next fetch returns an error(due to leadership change) and new leader info is returned, as introduced
        // in KIP-951. The new leader is a new node, id = 999. For tp1 fetch returns with no error.
        LinkedHashMap<TopicIdPartition, FetchResponseData.PartitionData> partitions = new LinkedHashMap<>();
        Node newNode = new Node(999, "newnode", 999, "newrack");
        FetchResponseData.PartitionData tp0Data = new FetchResponseData.PartitionData()
            .setPartitionIndex(tidp0.topicPartition().partition())
            .setErrorCode(error.code());
        tp0Data.currentLeader().setLeaderId(newNode.id());
        int tp0NewLeaderEpoch = validLeaderEpoch + 100;
        tp0Data.currentLeader().setLeaderEpoch(tp0NewLeaderEpoch);
        partitions.put(tidp0, tp0Data);
        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, singletonList(newNode)), nodeId0);
        networkClientDelegate.poll(time.timer(0));
        partitionRecords = fetchRecords();
        assertFalse(partitionRecords.containsKey(tp0));
        assertTrue(partitionRecords.containsKey(tp1));

        // Validate metadata is changed, as previous FetchResponse had new leader info for tp0.
        assertNotEquals(startingClusterMetadata, metadata.fetch());
        // Validate new-node(id=999) is part of the metadata
        assertEquals(newNode, metadata.fetch().nodeById(999));
        // Validate metadata returns the new leader info for tp0.
        Metadata.LeaderAndEpoch currentLeaderTp0 = metadata.currentLeader(tp0);
        assertEquals(Optional.of(newNode), currentLeaderTp0.leader);
        assertEquals(Optional.of(tp0NewLeaderEpoch), currentLeaderTp0.epoch);

        // Validate metadata-update is requested due to the leadership-error for 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 is ok.
        assertEquals(Optional.of(nodeId0.id()),
            subscriptions.preferredReadReplica(tp1, time.milliseconds()));

        // Validate subscription is valid & fetch-able, and points to the new leader.
        assertTrue(subscriptions.isFetchable(tp0));
        Metadata.LeaderAndEpoch currentLeader = subscriptions.position(tp0).currentLeader;
        assertEquals(newNode.id(), currentLeader.leader.get().id());
        assertEquals(tp0NewLeaderEpoch, currentLeader.epoch.get());

        // Validate subscription is still valid & fetch-able for tp1.
        assertTrue(subscriptions.isFetchable(tp1));
    }
Test the scenario that FetchResponse returns with an error indicating leadership change for the partition, along with
new leader info(defined in KIP-951).
clients/src/test/java/org/apache/kafka/clients/consumer/internals/FetchRequestManagerTest.java