def _handle_list_offsets_response(self, future, response):
        """Callback for the response of the ListOffsets api call

        Arguments:
            future (Future): the future to update based on response
            response (ListOffsetsResponse): response from the server

        Raises:
            AssertionError: if response does not match partition
        """
        fetched_offsets = dict()
        partitions_to_retry = set()
        unauthorized_topics = set()
        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 ListOffsetsResponse with one offset'
                        if not offsets:
                            offset = UNKNOWN_OFFSET
                        else:
                            offset = offsets[0]
                        timestamp = None
                        leader_epoch = -1
                    elif response.API_VERSION <= 3:
                        timestamp, offset = partition_info[2:]
                        leader_epoch = -1
                    else:
                        timestamp, offset, leader_epoch = partition_info[2:]
                    log.debug("Handling ListOffsetsResponse response for %s. "
                              "Fetched offset %s, timestamp %s, leader_epoch %s",
                              partition, offset, timestamp, leader_epoch)
                    if offset != UNKNOWN_OFFSET:
                        fetched_offsets[partition] = OffsetAndTimestamp(offset, timestamp, leader_epoch)
                elif error_type is Errors.UnsupportedForMessageFormatError:
                    # The message format on the broker side is before 0.10.0, which means it does not
                    # support timestamps. We treat this case the same as if we weren't able to find an
                    # offset corresponding to the requested timestamp and leave it out of the result.
                    log.debug("Cannot search by timestamp for partition %s because the"
                              " message format version is before 0.10.0", partition)
                elif error_type in (Errors.NotLeaderForPartitionError,
                                    Errors.ReplicaNotAvailableError,
                                    Errors.KafkaStorageError):
                    log.debug("Attempt to fetch offsets for partition %s failed due"
                              " to %s, retrying.", error_type.__name__, partition)
                    partitions_to_retry.add(partition)
                elif error_type is Errors.UnknownTopicOrPartitionError:
                    log.warning("Received unknown topic or partition error in ListOffsets "
                                "request for partition %s. The topic/partition " +
                                "may not exist or the user may not have Describe access "
                                "to it.", partition)
                    partitions_to_retry.add(partition)
                elif error_type is Errors.TopicAuthorizationFailedError:
                    unauthorized_topics.add(topic)
                else:
                    log.warning("Attempt to fetch offsets for partition %s failed due to:"
                                " %s", partition, error_type.__name__)
                    partitions_to_retry.add(partition)
        if unauthorized_topics:
            future.failure(Errors.TopicAuthorizationFailedError(unauthorized_topics))
        else:
            future.success((fetched_offsets, partitions_to_retry))
Callback for the response of the ListOffsets api call

        Arguments:
            future (Future): the future to update based on response
            response (ListOffsetsResponse): response from the server

        Raises:
            AssertionError: if response does not match partition
        
kafka/consumer/fetcher.py