private void testOffsetValidationWithGivenEpochOffset(int leaderEpoch,
                                                          long endOffset,
                                                          OffsetResetStrategy offsetResetStrategy) {
        buildFetcher(offsetResetStrategy);
        assignFromUser(singleton(tp0));

        Map<String, Integer> partitionCounts = new HashMap<>();
        partitionCounts.put(tp0.topic(), 4);

        final int epochOne = 1;
        final long initialOffset = 5;

        metadata.updateWithCurrentRequestVersion(RequestTestUtils.metadataUpdateWithIds("dummy", 1,
            Collections.emptyMap(), partitionCounts, tp -> epochOne, topicIds), false, 0L);

        // Offset validation requires OffsetForLeaderEpoch request v3 or higher
        Node node = metadata.fetch().nodes().get(0);
        apiVersions.update(node.idString(), NodeApiVersions.create());

        Metadata.LeaderAndEpoch leaderAndEpoch = new Metadata.LeaderAndEpoch(metadata.currentLeader(tp0).leader, Optional.of(epochOne));
        subscriptions.seekUnvalidated(tp0, new SubscriptionState.FetchPosition(initialOffset, Optional.of(epochOne), leaderAndEpoch));

        offsetFetcher.validatePositionsIfNeeded();

        consumerClient.poll(time.timer(Duration.ZERO));
        assertTrue(subscriptions.awaitingValidation(tp0));
        assertTrue(client.hasInFlightRequests());

        client.respond(
            offsetsForLeaderEpochRequestMatcher(tp0),
            prepareOffsetsForLeaderEpochResponse(tp0, leaderEpoch, endOffset));
        consumerClient.poll(time.timer(Duration.ZERO));

        if (offsetResetStrategy == OffsetResetStrategy.NONE) {
            LogTruncationException thrown =
                assertThrows(LogTruncationException.class, () -> offsetFetcher.validatePositionsIfNeeded());
            assertEquals(singletonMap(tp0, initialOffset), thrown.offsetOutOfRangePartitions());

            if (endOffset == UNDEFINED_EPOCH_OFFSET || leaderEpoch == UNDEFINED_EPOCH) {
                assertEquals(Collections.emptyMap(), thrown.divergentOffsets());
            } else {
                OffsetAndMetadata expectedDivergentOffset = new OffsetAndMetadata(
                    endOffset, Optional.of(leaderEpoch), "");
                assertEquals(singletonMap(tp0, expectedDivergentOffset), thrown.divergentOffsets());
            }
            assertTrue(subscriptions.awaitingValidation(tp0));
        } else {
            offsetFetcher.validatePositionsIfNeeded();
            assertFalse(subscriptions.awaitingValidation(tp0));
        }
    }
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