docstr: Callback for the response of the list offset call above.
  
  Arguments:
      future (Future): the future to update based on response
      response (OffsetResponse): response from the server
  
  Raises:
      AssertionError: if response does not match partition
code: def _handle_offset_response(self, future, response):
      """Callback for the response of the list offset call above.
  
      Arguments:
          future (Future): the future to update based on response
          response (OffsetResponse): response from the server
  
      Raises:
          AssertionError: if response does not match partition
      """
      timestamp_offset_map = {}
      for topic, part_data in response.topics:
          for partition_info in part_data:
              partition, error_code = partition_info[:2]
              partition = TopicPartition(topic, partition)
              error_type = Errors.for_code(error_code)
              if error_type is Errors.NoError:
                  if response.API_VERSION == 0:
                      offsets = partition_info[2]
                      assert (
                          len(offsets) <= 1
                      ), "Expected OffsetResponse with one offset"
                      if not offsets:
                          offset = UNKNOWN_OFFSET
                      else:
                          offset = offsets[0]
                      log.debug(
                          "Handling v0 ListOffsetResponse response for %s. "
                          "Fetched offset %s",
                          partition,
                          offset,
                      )
                      if offset != UNKNOWN_OFFSET:
                          timestamp_offset_map[partition] = (
                              offset,
                              None,
                          )
                  else:
                      timestamp, offset = partition_info[2:]
                      log.debug(
                          "Handling ListOffsetResponse response for %s. "
                          "Fetched offset %s, timestamp %s",
                          partition,
                          offset,
                          timestamp,
                      )
                      if offset != UNKNOWN_OFFSET:
                          timestamp_offset_map[partition] = (
                              offset,
                              timestamp,
                          )
              elif (
                  error_type is Errors.UnsupportedForMessageFormatError
              ):
                  # The message format on the broker side is before 0.10.0,
                  # we simply put None in the response.
                  log.debug(
                      "Cannot search by timestamp for partition %s because the"
                      " message format version is before 0.10.0",
                      partition,
                  )
              elif error_type is Errors.NotLeaderForPartitionError:
                  log.debug(
                      "Attempt to fetch offsets for partition %s failed due"
                      " to obsolete leadership information, retrying.",
                      partition,
                  )
                  future.failure(error_type(partition))
                  return
              elif error_type is Errors.UnknownTopicOrPartitionError:
                  log.warning(
                      "Received unknown topic or partition error in ListOffset "
                      "request for partition %s. The topic/partition "
                      + "may not exist or the user may not have Describe access "
                      "to it.",
                      partition,
                  )
                  future.failure(error_type(partition))
                  return
              else:
                  log.warning(
                      "Attempt to fetch offsets for partition %s failed due to:"
                      " %s",
                      partition,
                      error_type,
                  )
                  future.failure(error_type(partition))
                  return
      if not future.is_done:
          future.success(timestamp_offset_map)
  
imports: from __future__ import absolute_import
  import collections
  import copy
  import logging
  import random
  import sys
  import time
  from kafka.vendor import six
  import kafka.errors
  from kafka.future import Future
  from kafka.metrics.stats import Avg, Count, Max, Rate
  from kafka.protocol.fetch import FetchRequest
  from kafka.protocol.offset import OffsetRequest, OffsetResetStrategy, UNKNOWN_OFFSET
  from kafka.record import MemoryRecords
  from kafka.serializer import Deserializer
  from kafka.structs import TopicPartition, OffsetAndTimestamp
label_desc: Use image node_id 22 for calling the Fetcher obj's underlying member method code with example usage: obj._handle_offset_response(future, response) without return types