'Fetch the committed offsets for a set of partitions.
This is a non-blocking call. The returned future can be polled to get
the actual offsets returned from the broker.
Arguments:
partitions (list of TopicPartition): the partitions to fetch
Returns:
Future: resolves to dict of offsets: {TopicPartition: int}'

def _send_offset_fetch_request(self, partitions):

    assert (self.config['api_version'] >= (0, 8, 1)), 'Unsupported    Broker    API'
    assert all(map((lambda k: isinstance(k, TopicPartition)), partitions))
    if (not partitions):
        return Future().success({})
    elif self.coordinator_unknown():
        return Future().failure(Errors.GroupCoordinatorNotAvailableError)
    node_id = self.coordinator_id
    if (not self._client.ready(node_id)):
        log.debug('Node    %s    not    ready    --    failing    offset    fetch    request', node_id)
        return Future().failure(Errors.NodeNotReadyError)
    log.debug('Group    %s    fetching    committed    offsets    for    partitions:    %s', self.group_id, partitions)
    topic_partitions = collections.defaultdict(set)
    for tp in partitions:
        topic_partitions[tp.topic].add(tp.partition)
    if (self.config['api_version'] >= (0, 8, 2)):
        request = OffsetFetchRequest[1](self.group_id, list(topic_partitions.items()))
    else:
        request = OffsetFetchRequest[0](self.group_id, list(topic_partitions.items()))
    future = Future()
    _f = self._client.send(node_id, request)
    _f.add_callback(self._handle_offset_fetch_response, future)
    _f.add_errback(self._failed_request, node_id, request, future)
    return future
