@Test
    public void testOffsetValidationSkippedForOldBroker() {
        // Old brokers may require CLUSTER permission to use the OffsetForLeaderEpoch API,
        // so we should skip offset validation and not send the request.
        IsolationLevel isolationLevel = IsolationLevel.READ_UNCOMMITTED;
        int maxPollRecords = Integer.MAX_VALUE;
        long metadataExpireMs = Long.MAX_VALUE;
        OffsetResetStrategy offsetResetStrategy = OffsetResetStrategy.EARLIEST;
        int minBytes = 1;
        int maxBytes = Integer.MAX_VALUE;
        int maxWaitMs = 0;
        int fetchSize = 1000;

        MetricConfig metricConfig = new MetricConfig();
        LogContext logContext = new LogContext();
        SubscriptionState subscriptionState = new SubscriptionState(logContext, offsetResetStrategy);

        buildFetcher(metricConfig, isolationLevel, metadataExpireMs, subscriptionState, logContext);

        FetchMetricsRegistry metricsRegistry = new FetchMetricsRegistry(metricConfig.tags().keySet(), "consumertest-group");
        FetchConfig fetchConfig = new FetchConfig(
                minBytes,
                maxBytes,
                maxWaitMs,
                fetchSize,
                maxPollRecords,
                true, // check crc
                CommonClientConfigs.DEFAULT_CLIENT_RACK,
                isolationLevel);
        Fetcher<byte[], byte[]> fetcher = new Fetcher<>(
                logContext,
                consumerClient,
                metadata,
                subscriptions,
                fetchConfig,
                new Deserializers<>(new ByteArrayDeserializer(), new ByteArrayDeserializer()),
                new FetchMetricsManager(metrics, metricsRegistry),
                time,
                apiVersions);

        assignFromUser(singleton(tp0));

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

        final int epochOne = 1;
        final int epochTwo = 2;

        // Start with metadata, epoch=1
        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(
            ApiKeys.OFFSET_FOR_LEADER_EPOCH.id, (short) 0, (short) 2));

        {
            // Seek with a position and leader+epoch
            Metadata.LeaderAndEpoch leaderAndEpoch = new Metadata.LeaderAndEpoch(
                    metadata.currentLeader(tp0).leader, Optional.of(epochOne));
            subscriptions.seekUnvalidated(tp0, new SubscriptionState.FetchPosition(0, Optional.of(epochOne), leaderAndEpoch));

            // Update metadata to epoch=2, enter validation
            metadata.updateWithCurrentRequestVersion(RequestTestUtils.metadataUpdateWithIds("dummy", 1,
                    Collections.emptyMap(), partitionCounts, tp -> epochTwo, topicIds), false, 0L);
            offsetFetcher.validatePositionsIfNeeded();

            // Offset validation is skipped
            assertFalse(subscriptions.awaitingValidation(tp0));
        }

        {
            // Seek with a position and leader+epoch
            Metadata.LeaderAndEpoch leaderAndEpoch = new Metadata.LeaderAndEpoch(
                    metadata.currentLeader(tp0).leader, Optional.of(epochOne));
            subscriptions.seekUnvalidated(tp0, new SubscriptionState.FetchPosition(0, Optional.of(epochOne), leaderAndEpoch));

            // Update metadata to epoch=2, enter validation
            metadata.updateWithCurrentRequestVersion(RequestTestUtils.metadataUpdateWithIds("dummy", 1,
                    Collections.emptyMap(), partitionCounts, tp -> epochTwo, topicIds), false, 0L);

            // Subscription should not stay in AWAITING_VALIDATION in prepareFetchRequest
            offsetFetcher.validatePositionsOnMetadataChange();
            assertEquals(1, fetcher.sendFetches());
            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