'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'

def _handle_offset_response(self, future, response):

    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):
                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.warn(('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)
