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:
            if self.config['api_version'] >= (0, 8, 2):
                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 # pylint: disable-msg=raising-bad-type

            time.sleep(self.config['retry_backoff_ms'] / 1000.0)
Commit specific offsets synchronously.
dataset/ETHPy150Open dpkp/kafka-python/kafka/coordinator/consumer.py/ConsumerCoordinator.commit_offsets_sync