@Test
    public void testListOffsetNoUpdateMissingEpoch() {
        buildFetcher();

        // Set up metadata with no leader epoch
        subscriptions.assignFromUser(singleton(tp0));
        MetadataResponse metadataWithNoLeaderEpochs = RequestTestUtils.metadataUpdateWithIds(
                "kafka-cluster", 1, Collections.emptyMap(), singletonMap(topicName, 4), tp -> null, topicIds);
        client.updateMetadata(metadataWithNoLeaderEpochs);

        // Return a ListOffsets response with leaderEpoch=1, we should ignore it
        subscriptions.requestOffsetReset(tp0, OffsetResetStrategy.LATEST);
        client.prepareResponse(listOffsetRequestMatcher(ListOffsetsRequest.LATEST_TIMESTAMP),
                listOffsetResponse(tp0, Errors.NONE, 1L, 5L, 1));
        offsetFetcher.resetPositionsIfNeeded();
        consumerClient.pollNoWakeup();

        // Reset should be satisfied and no metadata update requested
        assertFalse(subscriptions.isOffsetResetNeeded(tp0));
        assertFalse(metadata.updateRequested());
        assertFalse(metadata.lastSeenLeaderEpoch(tp0).isPresent());
    }
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