class_docstr: This class manages the coordination process with the consumer coordinator.
code: def _handle_offset_fetch_response(self, future, response):
      offsets = {}
      for topic, partitions in response.topics:
          for partition, offset, metadata, error_code in partitions:
              tp = TopicPartition(topic, partition)
              error_type = Errors.for_code(error_code)
              if error_type is not Errors.NoError:
                  error = error_type()
                  log.debug(
                      "Group %s failed to fetch offset for partition"
                      " %s: %s",
                      self.group_id,
                      tp,
                      error,
                  )
                  if error_type is Errors.GroupLoadInProgressError:
                      # just retry
                      future.failure(error)
                  elif error_type is Errors.NotCoordinatorForGroupError:
                      # re-discover the coordinator and retry
                      self.coordinator_dead(error_type())
                      future.failure(error)
                  elif (
                      error_type is Errors.UnknownTopicOrPartitionError
                  ):
                      log.warning(
                          "OffsetFetchRequest -- unknown topic %s"
                          " (have you committed any offsets yet?)",
                          topic,
                      )
                      continue
                  else:
                      log.error(
                          "Unknown error fetching offsets for %s: %s",
                          tp,
                          error,
                      )
                      future.failure(error)
                  return
              elif offset >= 0:
                  # record the position with the offset
                  # (-1 indicates no committed offset to fetch)
                  offsets[tp] = OffsetAndMetadata(offset, metadata)
              else:
                  log.debug(
                      "Group %s has no committed offset for partition"
                      " %s",
                      self.group_id,
                      tp,
                  )
      future.success(offsets)
  
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 26 for calling the ConsumerCoordinator obj's underlying member method code with example usage: obj._handle_offset_fetch_response(future, response) without return types