'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'

def _send_offset_requests(self, timestamps):

    timestamps_by_node = collections.defaultdict(dict)
    for (partition, timestamp) in six.iteritems(timestamps):
        node_id = self._client.cluster.leader_for_partition(partition)
        if (node_id is None):
            self._client.add_topic(partition.topic)
            log.debug('Partition    %s    is    unknown    for    fetching    offset,    wait    for    metadata    refresh', partition)
            return Future().failure(Errors.StaleMetadata(partition))
        elif (node_id == (-1)):
            log.debug('Leader    for    partition    %s    unavailable    for    fetching    offset,    wait    for    metadata    refresh', partition)
            return Future().failure(Errors.LeaderNotAvailableError(partition))
        else:
            timestamps_by_node[node_id][partition] = timestamp
    list_offsets_future = Future()
    responses = []
    node_count = len(timestamps_by_node)
    def on_success(value):
        responses.append(value)
        if (len(responses) == node_count):
            offsets = {}
            for r in responses:
                offsets.update(r)
            list_offsets_future.success(offsets)
    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_offset_request(node_id, timestamps)
        _f.add_callback(on_success)
        _f.add_errback(on_fail)
    return list_offsets_future
