@Test
    public void testGetOffsetsFencedLeaderEpoch() {
        buildFetcher();
        subscriptions.assignFromUser(singleton(tp0));
        client.updateMetadata(initialUpdateResponse);

        subscriptions.requestOffsetReset(tp0, OffsetResetStrategy.LATEST);

        client.prepareResponse(listOffsetResponse(Errors.FENCED_LEADER_EPOCH, 1L, 5L));
        offsetFetcher.resetPositionsIfNeeded();
        consumerClient.pollNoWakeup();

        assertTrue(subscriptions.isOffsetResetNeeded(tp0));
        assertFalse(subscriptions.isFetchable(tp0));
        assertFalse(subscriptions.hasValidPosition(tp0));
        assertEquals(0L, metadata.timeToNextUpdate(time.milliseconds()));
    }
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