code: 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()))
          self._invoke_completed_offset_commit_callbacks()
          if not offsets:
              return
  
          while True:
              self.ensure_coordinator_ready()
  
              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)
docs: 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