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

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

        // Reset offsets to trigger ListOffsets call
        subscriptions.requestOffsetReset(tp0, OffsetResetStrategy.LATEST);

        // Now we see a ListOffsets with leaderEpoch=2 epoch, we trigger a metadata update
        client.prepareResponse(listOffsetRequestMatcher(ListOffsetsRequest.LATEST_TIMESTAMP, 1),
                listOffsetResponse(tp0, Errors.NONE, 1L, 5L, 2));
        offsetFetcher.resetPositionsIfNeeded();
        consumerClient.pollNoWakeup();

        assertFalse(subscriptions.isOffsetResetNeeded(tp0));
        assertTrue(metadata.updateRequested());
        assertOptional(metadata.lastSeenLeaderEpoch(tp0), epoch -> assertEquals((long) epoch, 2));
    }
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