class_docstr: This class manages the coordination process with the consumer coordinator.
docstr: Commit offsets for the specified list of topics and partitions.
  
  This is a non-blocking call which returns a request future that can be
  polled in the case of a synchronous commit or ignored in the
  asynchronous case.
  
  Arguments:
      offsets (dict of {TopicPartition: OffsetAndMetadata}): what should
          be committed
  
  Returns:
      Future: indicating whether the commit was successful or not
docstr_tok: ["Commit","offsets","for","the","specified","list","of","topics","and","partitions",".","This","is","a","non-blocking","call","which","returns","a","request","future","that","can","be","polled","in","the","case","of","a","synchronous","commit","or","ignored","in","the","asynchronous","case",".","Arguments",":","offsets","(","dict","of","{","TopicPartition",":","OffsetAndMetadata","}",")",":","what","should","be","committed","Returns",":","Future",":","indicating","whether","the","commit","was","successful","or","not"]
code: def _send_offset_commit_request(self, offsets):
      """Commit offsets for the specified list of topics and partitions.
  
      This is a non-blocking call which returns a request future that can be
      polled in the case of a synchronous commit or ignored in the
      asynchronous case.
  
      Arguments:
          offsets (dict of {TopicPartition: OffsetAndMetadata}): what should
              be committed
  
      Returns:
          Future: indicating whether the commit was successful or not
      """
      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:
          log.debug("No offsets to commit")
          return Future().success(None)
  
      node_id = self.coordinator()
      if node_id is None:
          return Future().failure(
              Errors.GroupCoordinatorNotAvailableError
          )
  
      # create the offset commit request
      offset_data = collections.defaultdict(dict)
      for tp, offset in six.iteritems(offsets):
          offset_data[tp.topic][tp.partition] = offset
  
      if self._subscription.partitions_auto_assigned():
          generation = self.generation()
      else:
          generation = Generation.NO_GENERATION
  
      # if the generation is None, we are not part of an active group
      # (and we expect to be). The only thing we can do is fail the commit
      # and let the user rejoin the group in poll()
      if self.config["api_version"] >= (0, 9) and generation is None:
          return Future().failure(Errors.CommitFailedError())
  
      if self.config["api_version"] >= (0, 9):
          request = OffsetCommitRequest[2](
              self.group_id,
              generation.generation_id,
              generation.member_id,
              OffsetCommitRequest[2].DEFAULT_RETENTION_TIME,
              [
                  (
                      topic,
                      [
                          (partition, offset.offset, offset.metadata)
                          for partition, offset in six.iteritems(
                              partitions
                          )
                      ],
                  )
                  for topic, partitions in six.iteritems(offset_data)
              ],
          )
      elif self.config["api_version"] >= (0, 8, 2):
          request = OffsetCommitRequest[1](
              self.group_id,
              -1,
              "",
              [
                  (
                      topic,
                      [
                          (
                              partition,
                              offset.offset,
                              -1,
                              offset.metadata,
                          )
                          for partition, offset in six.iteritems(
                              partitions
                          )
                      ],
                  )
                  for topic, partitions in six.iteritems(offset_data)
              ],
          )
      elif self.config["api_version"] >= (0, 8, 1):
          request = OffsetCommitRequest[0](
              self.group_id,
              [
                  (
                      topic,
                      [
                          (partition, offset.offset, offset.metadata)
                          for partition, offset in six.iteritems(
                              partitions
                          )
                      ],
                  )
                  for topic, partitions in six.iteritems(offset_data)
              ],
          )
  
      log.debug(
          "Sending offset-commit request with %s for group %s to %s",
          offsets,
          self.group_id,
          node_id,
      )
  
      future = Future()
      _f = self._client.send(node_id, request)
      _f.add_callback(
          self._handle_offset_commit_response,
          offsets,
          future,
          time.time(),
      )
      _f.add_errback(self._failed_request, node_id, request, future)
      return future
  
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 23 for calling the ConsumerCoordinator obj's underlying member method code with example usage: obj._send_offset_commit_request(offsets) and returns: future, Future, Future, Future