class_docstr: This class manages the coordination process with the consumer coordinator.
docstr: 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
docstr_tok: ["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"]
code: def commit_offsets_sync(self, offsets):
      """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
      """
      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)
  
imports: from __future__ import absolute_import, division
  import collections
  import copy
  import functools
  import logging
  import time
  from kafka.vendor import six
  from kafka.coordinator.base import BaseCoordinator, Generation
  from kafka.coordinator.assignors.range import RangePartitionAssignor
  from kafka.coordinator.assignors.roundrobin import RoundRobinPartitionAssignor
  from kafka.coordinator.assignors.sticky.sticky_assignor import StickyPartitionAssignor
  from kafka.coordinator.protocol import ConsumerProtocol
  import kafka.errors
  from kafka.future import Future
  from kafka.metrics import AnonMeasurable
  from kafka.metrics.stats import Avg, Count, Max, Rate
  from kafka.protocol.commit import OffsetCommitRequest, OffsetFetchRequest
  from kafka.structs import OffsetAndMetadata, TopicPartition
  from kafka.util import WeakMethod
label_desc: Use image node_id 21 for calling the ConsumerCoordinator obj's underlying member method code with example usage: obj.commit_offsets_sync(offsets) and returns: future