public CompletableFuture<Map<TopicPartition, OffsetAndTimestampInternal>> fetchOffsets(
            Map<TopicPartition, Long> timestampsToSearch,
            boolean requireTimestamps) {
        if (timestampsToSearch.isEmpty()) {
            return CompletableFuture.completedFuture(Collections.emptyMap());
        }
        metadata.addTransientTopics(OffsetFetcherUtils.topicsForPartitions(timestampsToSearch.keySet()));
        ListOffsetsRequestState listOffsetsRequestState = new ListOffsetsRequestState(
                timestampsToSearch,
                requireTimestamps,
                offsetFetcherUtils,
                isolationLevel);
        listOffsetsRequestState.globalResult.whenComplete((result, error) -> {
            metadata.clearTransientTopics();
            if (error != null) {
                log.debug("Fetch offsets completed with error for partitions and timestamps {}.",
                        timestampsToSearch, error);
            } else {
                log.debug("Fetch offsets completed successfully for partitions and timestamps {}." +
                        " Result {}", timestampsToSearch, result);
            }
        });

        prepareFetchOffsetsRequests(timestampsToSearch, requireTimestamps, listOffsetsRequestState);
        return listOffsetsRequestState.globalResult.thenApply(
                result -> OffsetFetcherUtils.buildOffsetsForTimeInternalResult(
                        timestampsToSearch,
                        result.fetchedOffsets));
    }
Retrieve offsets for the given partitions and timestamp. For each partition, this will
retrieve the offset of the first message whose timestamp is greater than or equals to the
target timestamp.

@param timestampsToSearch Partitions and target timestamps to get offsets for
@param requireTimestamps  True if this should fail with an UnsupportedVersionException if the
                          broker does not support fetching precise timestamps for offsets
@return Future containing the map of {@link TopicPartition} and {@link OffsetAndTimestamp}
found .The future will complete when the requests responses are received and
processed, following a call to {@link #poll(long)}
clients/src/main/java/org/apache/kafka/clients/consumer/internals/OffsetsRequestManager.java