@Test
    public void testGetOffsetByTimeWithPartitionsRetryCouldTriggerMetadataUpdate() {
        List<Errors> retriableErrors = Arrays.asList(Errors.NOT_LEADER_OR_FOLLOWER,
            Errors.REPLICA_NOT_AVAILABLE, Errors.KAFKA_STORAGE_ERROR, Errors.OFFSET_NOT_AVAILABLE,
            Errors.LEADER_NOT_AVAILABLE, Errors.FENCED_LEADER_EPOCH, Errors.UNKNOWN_LEADER_EPOCH);

        final int newLeaderEpoch = 3;
        MetadataResponse updatedMetadata = RequestTestUtils.metadataUpdateWithIds("dummy", 3,
            singletonMap(topicName, Errors.NONE), singletonMap(topicName, 4), tp -> newLeaderEpoch, topicIds);

        Node originalLeader = initialUpdateResponse.buildCluster().leaderFor(tp1);
        Node newLeader = updatedMetadata.buildCluster().leaderFor(tp1);
        assertNotEquals(originalLeader, newLeader);

        for (Errors retriableError : retriableErrors) {
            buildFetcher();

            subscriptions.assignFromUser(mkSet(tp0, tp1));
            client.updateMetadata(initialUpdateResponse);

            final long fetchTimestamp = 10L;
            ListOffsetsPartitionResponse tp0NoError = new ListOffsetsPartitionResponse()
                .setPartitionIndex(tp0.partition())
                .setErrorCode(Errors.NONE.code())
                .setTimestamp(fetchTimestamp)
                .setOffset(4L);
            List<ListOffsetsTopicResponse> topics = Collections.singletonList(
                    new ListOffsetsTopicResponse()
                        .setName(tp0.topic())
                        .setPartitions(Arrays.asList(
                                tp0NoError,
                                new ListOffsetsPartitionResponse()
                                    .setPartitionIndex(tp1.partition())
                                    .setErrorCode(retriableError.code())
                                    .setTimestamp(ListOffsetsRequest.LATEST_TIMESTAMP)
                                    .setOffset(-1L))));
            ListOffsetsResponseData data = new ListOffsetsResponseData()
                    .setThrottleTimeMs(0)
                    .setTopics(topics);

            client.prepareResponseFrom(body -> {
                boolean isListOffsetRequest = body instanceof ListOffsetsRequest;
                if (isListOffsetRequest) {
                    ListOffsetsRequest request = (ListOffsetsRequest) body;
                    List<ListOffsetsTopic> expectedTopics = Collections.singletonList(
                            new ListOffsetsTopic()
                                .setName(tp0.topic())
                                .setPartitions(Arrays.asList(
                                        new ListOffsetsPartition()
                                            .setPartitionIndex(tp1.partition())
                                            .setTimestamp(fetchTimestamp)
                                            .setCurrentLeaderEpoch(ListOffsetsResponse.UNKNOWN_EPOCH),
                                        new ListOffsetsPartition()
                                            .setPartitionIndex(tp0.partition())
                                            .setTimestamp(fetchTimestamp)
                                            .setCurrentLeaderEpoch(ListOffsetsResponse.UNKNOWN_EPOCH))));
                    return request.topics().equals(expectedTopics);
                } else {
                    return false;
                }
            }, new ListOffsetsResponse(data), originalLeader);

            client.prepareMetadataUpdate(updatedMetadata);

            // If the metadata wasn't updated before retrying, the fetcher would consult the original leader and hit a NOT_LEADER exception.
            // We will count the answered future response in the end to verify if this is the case.
            List<ListOffsetsTopicResponse> topicsWithFatalError = Collections.singletonList(
                    new ListOffsetsTopicResponse()
                        .setName(tp0.topic())
                        .setPartitions(Arrays.asList(
                                tp0NoError,
                                new ListOffsetsPartitionResponse()
                                    .setPartitionIndex(tp1.partition())
                                    .setErrorCode(Errors.NOT_LEADER_OR_FOLLOWER.code())
                                    .setTimestamp(ListOffsetsRequest.LATEST_TIMESTAMP)
                                    .setOffset(-1L))));
            ListOffsetsResponseData dataWithFatalError = new ListOffsetsResponseData()
                    .setThrottleTimeMs(0)
                    .setTopics(topicsWithFatalError);
            client.prepareResponseFrom(new ListOffsetsResponse(dataWithFatalError), originalLeader);

            // The request to new leader must only contain one partition tp1 with error.
            client.prepareResponseFrom(body -> {
                boolean isListOffsetRequest = body instanceof ListOffsetsRequest;
                if (isListOffsetRequest) {
                    ListOffsetsRequest request = (ListOffsetsRequest) body;

                    ListOffsetsTopic requestTopic = request.topics().get(0);
                    ListOffsetsPartition expectedPartition = new ListOffsetsPartition()
                            .setPartitionIndex(tp1.partition())
                            .setTimestamp(fetchTimestamp)
                            .setCurrentLeaderEpoch(newLeaderEpoch);
                    return expectedPartition.equals(requestTopic.partitions().get(0));
                } else {
                    return false;
                }
            }, listOffsetResponse(tp1, Errors.NONE, fetchTimestamp, 5L), newLeader);

            Map<TopicPartition, OffsetAndTimestamp> offsetAndTimestampMap =
                offsetFetcher.offsetsForTimes(
                    Utils.mkMap(Utils.mkEntry(tp0, fetchTimestamp),
                    Utils.mkEntry(tp1, fetchTimestamp)), time.timer(Integer.MAX_VALUE));

            assertEquals(Utils.mkMap(
                Utils.mkEntry(tp0, new OffsetAndTimestamp(4L, fetchTimestamp)),
                Utils.mkEntry(tp1, new OffsetAndTimestamp(5L, fetchTimestamp))), offsetAndTimestampMap);

            // The NOT_LEADER exception future should not be cleared as we already refreshed the metadata before
            // first retry, thus never hitting.
            assertEquals(1, client.numAwaitingResponses());
        }
    }
Make sure the client behaves appropriately when receiving an exception for unavailable offsets
clients/src/test/java/org/apache/kafka/clients/consumer/internals/OffsetFetcherTest.java