'Commit specific offsets asynchronously.
Arguments:
offsets (dict {TopicPartition: OffsetAndMetadata}): what to commit
callback (callable, optional): called as callback(offsets, response)
response will be either an Exception or a OffsetCommitResponse
struct. This callback can be used to trigger custom actions when
a commit request completes.
Returns:
Future: indicating whether the commit was successful or not'

def commit_offsets_async(self, offsets, callback=None):

    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 (callback is None):
        callback = self.config['default_offset_commit_callback']
    self._subscription.needs_fetch_committed_offsets = True
    future = self._send_offset_commit_request(offsets)
    future.add_both(callback, offsets)
    return future
