private MockClient.RequestMatcher listOffsetRequestMatcher(final long timestamp, final int leaderEpoch) {
        // matches any list offset request with the provided timestamp
        return body -> {
            ListOffsetsRequest req = (ListOffsetsRequest) body;
            ListOffsetsTopic topic = req.topics().get(0);
            ListOffsetsPartition partition = topic.partitions().get(0);
            return tp0.topic().equals(topic.name())
                    && tp0.partition() == partition.partitionIndex()
                    && timestamp == partition.timestamp()
                    && leaderEpoch == partition.currentLeaderEpoch();
        };
    }
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