'Commit offsets for the specified list of topics and partitions.
This is a non-blocking call which returns a request future that can be
polled in the case of a synchronous commit or ignored in the
asynchronous case.
Arguments:
offsets (dict of {TopicPartition: OffsetAndMetadata}): what should
be committed
Returns:
Future: indicating whether the commit was successful or not'

def _send_offset_commit_request(self, offsets):

    assert (self.config['api_version'] >= (0, 8, 1)), 'Unsupported    Broker    API'
    assert all(map((lambda k: isinstance(k, TopicPartition)), offsets))
    assert all(map((lambda v: isinstance(v, OffsetAndMetadata)), offsets.values()))
    if (not offsets):
        log.debug('No    offsets    to    commit')
        return Future().success(True)
    elif self.coordinator_unknown():
        return Future().failure(Errors.GroupCoordinatorNotAvailableError)
    node_id = self.coordinator_id
    offset_data = collections.defaultdict(dict)
    for (tp, offset) in six.iteritems(offsets):
        offset_data[tp.topic][tp.partition] = offset
    if (self.config['api_version'] >= (0, 9)):
        request = OffsetCommitRequest[2](self.group_id, self.generation, self.member_id, OffsetCommitRequest[2].DEFAULT_RETENTION_TIME, [(topic, [(partition, offset.offset, offset.metadata) for (partition, offset) in six.iteritems(partitions)]) for (topic, partitions) in six.iteritems(offset_data)])
    elif (self.config['api_version'] >= (0, 8, 2)):
        request = OffsetCommitRequest[1](self.group_id, (-1), '', [(topic, [(partition, offset.offset, (-1), offset.metadata) for (partition, offset) in six.iteritems(partitions)]) for (topic, partitions) in six.iteritems(offset_data)])
    elif (self.config['api_version'] >= (0, 8, 1)):
        request = OffsetCommitRequest[0](self.group_id, [(topic, [(partition, offset.offset, offset.metadata) for (partition, offset) in six.iteritems(partitions)]) for (topic, partitions) in six.iteritems(offset_data)])
    log.debug('Sending    offset-commit    request    with    %s    for    group    %s    to    %s', offsets, self.group_id, node_id)
    future = Future()
    _f = self._client.send(node_id, request)
    _f.add_callback(self._handle_offset_commit_response, offsets, future, time.time())
    _f.add_errback(self._failed_request, node_id, request, future)
    return future
