class_docstr: This class manages the coordination process with the consumer coordinator.
code: def _handle_offset_commit_response(
      self, offsets, future, send_time, response
  ):
      # TODO look at adding request_latency_ms to response (like java kafka)
      self.consumer_sensors.commit_latency.record(
          (time.time() - send_time) * 1000
      )
      unauthorized_topics = set()
  
      for topic, partitions in response.topics:
          for partition, error_code in partitions:
              tp = TopicPartition(topic, partition)
              offset = offsets[tp]
  
              error_type = Errors.for_code(error_code)
              if error_type is Errors.NoError:
                  log.debug(
                      "Group %s committed offset %s for partition %s",
                      self.group_id,
                      offset,
                      tp,
                  )
                  if self._subscription.is_assigned(tp):
                      self._subscription.assignment[
                          tp
                      ].committed = offset
              elif error_type is Errors.GroupAuthorizationFailedError:
                  log.error(
                      "Not authorized to commit offsets for group %s",
                      self.group_id,
                  )
                  future.failure(error_type(self.group_id))
                  return
              elif error_type is Errors.TopicAuthorizationFailedError:
                  unauthorized_topics.add(topic)
              elif error_type in (
                  Errors.OffsetMetadataTooLargeError,
                  Errors.InvalidCommitOffsetSizeError,
              ):
                  # raise the error to the user
                  log.debug(
                      "OffsetCommit for group %s failed on partition %s"
                      " %s",
                      self.group_id,
                      tp,
                      error_type.__name__,
                  )
                  future.failure(error_type())
                  return
              elif error_type is Errors.GroupLoadInProgressError:
                  # just retry
                  log.debug(
                      "OffsetCommit for group %s failed: %s",
                      self.group_id,
                      error_type.__name__,
                  )
                  future.failure(error_type(self.group_id))
                  return
              elif error_type in (
                  Errors.GroupCoordinatorNotAvailableError,
                  Errors.NotCoordinatorForGroupError,
                  Errors.RequestTimedOutError,
              ):
                  log.debug(
                      "OffsetCommit for group %s failed: %s",
                      self.group_id,
                      error_type.__name__,
                  )
                  self.coordinator_dead(error_type())
                  future.failure(error_type(self.group_id))
                  return
              elif error_type in (
                  Errors.UnknownMemberIdError,
                  Errors.IllegalGenerationError,
                  Errors.RebalanceInProgressError,
              ):
                  # need to re-join group
                  error = error_type(self.group_id)
                  log.debug(
                      "OffsetCommit for group %s failed: %s",
                      self.group_id,
                      error,
                  )
                  self.reset_generation()
                  future.failure(Errors.CommitFailedError())
                  return
              else:
                  log.error(
                      "Group %s failed to commit partition %s at offset"
                      " %s: %s",
                      self.group_id,
                      tp,
                      offset,
                      error_type.__name__,
                  )
                  future.failure(error_type())
                  return
  
      if unauthorized_topics:
          log.error(
              "Not authorized to commit to topics %s for group %s",
              unauthorized_topics,
              self.group_id,
          )
          future.failure(
              Errors.TopicAuthorizationFailedError(unauthorized_topics)
          )
      else:
          future.success(None)
  
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 24 for calling the ConsumerCoordinator obj's underlying member method code with example usage: obj._handle_offset_commit_response(offsets, future, send_time, response) without return types