private boolean listOffsetMatchesExpectedReset(
        TopicPartition tp,
        OffsetResetStrategy strategy,
        AbstractRequest request
    ) {
        assertInstanceOf(ListOffsetsRequest.class, request);

        ListOffsetsRequest req = (ListOffsetsRequest) request;
        assertEquals(singleton(tp.topic()), req.data().topics().stream()
            .map(ListOffsetsTopic::name).collect(Collectors.toSet()));

        ListOffsetsTopic listTopic = req.data().topics().get(0);
        assertEquals(singleton(tp.partition()), listTopic.partitions().stream()
            .map(ListOffsetsPartition::partitionIndex).collect(Collectors.toSet()));

        ListOffsetsPartition listPartition = listTopic.partitions().get(0);
        if (strategy == OffsetResetStrategy.EARLIEST) {
            assertEquals(ListOffsetsRequest.EARLIEST_TIMESTAMP, listPartition.timestamp());
        } else if (strategy == OffsetResetStrategy.LATEST) {
            assertEquals(ListOffsetsRequest.LATEST_TIMESTAMP, listPartition.timestamp());
        }
        return true;
    }
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