def _send_list_offsets_requests(self, timestamps):
        """Fetch offsets for each partition in timestamps dict. This may send
        request to multiple nodes, based on who is Leader for partition.

        Arguments:
            timestamps (dict): {TopicPartition: int} mapping of fetching
                timestamps.

        Returns:
            Future: resolves to a mapping of retrieved offsets
        """
        timestamps_by_node = self._group_list_offset_requests(timestamps)
        if not timestamps_by_node:
            return Future().failure(Errors.StaleMetadata())

        # Aggregate results until we have all responses
        list_offsets_future = Future()
        fetched_offsets = dict()
        partitions_to_retry = set()
        remaining_responses = [len(timestamps_by_node)] # list for mutable / 2.7 hack

        def on_success(remaining_responses, value):
            remaining_responses[0] -= 1 # noqa: F823
            fetched_offsets.update(value[0])
            partitions_to_retry.update(value[1])
            if not remaining_responses[0] and not list_offsets_future.is_done:
                list_offsets_future.success((fetched_offsets, partitions_to_retry))

        def on_fail(err):
            if not list_offsets_future.is_done:
                list_offsets_future.failure(err)

        for node_id, timestamps in six.iteritems(timestamps_by_node):
            _f = self._send_list_offsets_request(node_id, timestamps)
            _f.add_callback(on_success, remaining_responses)
            _f.add_errback(on_fail)
        return list_offsets_future
Fetch offsets for each partition in timestamps dict. This may send
        request to multiple nodes, based on who is Leader for partition.

        Arguments:
            timestamps (dict): {TopicPartition: int} mapping of fetching
                timestamps.

        Returns:
            Future: resolves to a mapping of retrieved offsets
        
kafka/consumer/fetcher.py