private void testListOffsetsSendsIsolationLevel(IsolationLevel isolationLevel) {
        buildFetcher(isolationLevel);

        assignFromUser(singleton(tp0));
        subscriptions.requestOffsetReset(tp0, OffsetResetStrategy.LATEST);

        client.prepareResponse(body -> {
            ListOffsetsRequest request = (ListOffsetsRequest) body;
            return request.isolationLevel() == isolationLevel;
        }, listOffsetResponse(Errors.NONE, 1L, 5L));
        offsetFetcher.resetPositionsIfNeeded();
        consumerClient.pollNoWakeup();

        assertFalse(subscriptions.isOffsetResetNeeded(tp0));
        assertTrue(subscriptions.isFetchable(tp0));
        assertEquals(5, subscriptions.position(tp0).offset);
    }
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