docs: 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)