@Test
    public void testGetOffsetsIncludesLeaderEpoch() {
        buildFetcher();
        subscriptions.assignFromUser(singleton(tp0));

        client.updateMetadata(initialUpdateResponse);

        // Metadata update with leader epochs
        MetadataResponse metadataResponse = RequestTestUtils.metadataUpdateWithIds("dummy", 1,
                Collections.emptyMap(), Collections.singletonMap(topicName, 4), tp -> 99, topicIds);
        client.updateMetadata(metadataResponse);

        // Request latest offset
        subscriptions.requestOffsetReset(tp0);
        offsetFetcher.resetPositionsIfNeeded();

        // Check for epoch in outgoing request
        MockClient.RequestMatcher matcher = body -> {
            if (body instanceof ListOffsetsRequest) {
                ListOffsetsRequest offsetRequest = (ListOffsetsRequest) body;
                int epoch = offsetRequest.topics().get(0).partitions().get(0).currentLeaderEpoch();
                assertTrue(epoch != ListOffsetsResponse.UNKNOWN_EPOCH, "Expected Fetcher to set leader epoch in request");
                assertEquals(epoch, 99, "Expected leader epoch to match epoch from metadata update");
                return true;
            } else {
                fail("Should have seen ListOffsetRequest");
                return false;
            }
        };

        client.prepareResponse(matcher, listOffsetResponse(Errors.NONE, 1L, 5L));
        consumerClient.pollNoWakeup();
    }
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