@Test
    public void testPreferredReadReplicaOffsetError() {
        buildFetcher(new MetricConfig(), OffsetResetStrategy.EARLIEST, new BytesDeserializer(), new BytesDeserializer(),
                Integer.MAX_VALUE, IsolationLevel.READ_COMMITTED, Duration.ofMinutes(5).toMillis());

        subscriptions.assignFromUser(singleton(tp0));
        client.updateMetadata(RequestTestUtils.metadataUpdateWithIds(2, singletonMap(topicName, 4), tp -> validLeaderEpoch, topicIds, false));

        subscriptions.seek(tp0, 0);

        assertEquals(1, sendFetches());
        assertFalse(fetcher.hasCompletedFetches());

        client.prepareResponse(fullFetchResponse(tidp0, records, Errors.NONE, 100L,
                FetchResponse.INVALID_LAST_STABLE_OFFSET, 0, Optional.of(1)));
        networkClientDelegate.poll(time.timer(0));
        assertTrue(fetcher.hasCompletedFetches());

        fetchRecords();

        Node selected = fetcher.selectReadReplica(tp0, Node.noNode(), time.milliseconds());
        assertEquals(selected.id(), 1);

        assertEquals(1, sendFetches());
        assertFalse(fetcher.hasCompletedFetches());

        // Return an error, should unset the preferred read replica
        client.prepareResponse(fullFetchResponse(tidp0, records, Errors.OFFSET_OUT_OF_RANGE, 100L,
                FetchResponse.INVALID_LAST_STABLE_OFFSET, 0, Optional.empty()));
        networkClientDelegate.poll(time.timer(0));
        assertTrue(fetcher.hasCompletedFetches());

        fetchRecords();

        selected = fetcher.selectReadReplica(tp0, Node.noNode(), time.milliseconds());
        assertEquals(selected.id(), -1);
    }
Supplies parameters to {@link #testHandleFetchResponseError(Errors, long, boolean, boolean)}.
clients/src/test/java/org/apache/kafka/clients/consumer/internals/FetchRequestManagerTest.java