'Commit specific offsets synchronously.
This method will retry until the commit completes successfully or an
unrecoverable error is encountered.
Arguments:
offsets (dict {TopicPartition: OffsetAndMetadata}): what to commit
Raises error on failure'

def commit_offsets_sync(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):
        return
    while True:
        self.ensure_coordinator_known()
        future = self._send_offset_commit_request(offsets)
        self._client.poll(future=future)
        if future.succeeded():
            return future.value
        if (not future.retriable()):
            raise future.exception
        time.sleep((self.config['retry_backoff_ms'] / 1000.0))
