'The callback for fetch completion'

def _handle_fetch_response(self, request, send_time, response):

    total_bytes = 0
    total_count = 0
    recv_time = time.time()
    fetch_offsets = {}
    for (topic, partitions) in request.topics:
        for (partition, offset, _) in partitions:
            fetch_offsets[TopicPartition(topic, partition)] = offset
    random.shuffle(response.topics)
    for (topic, partitions) in response.topics:
        random.shuffle(partitions)
        for (partition, error_code, highwater, messages) in partitions:
            tp = TopicPartition(topic, partition)
            error_type = Errors.for_code(error_code)
            if (not self._subscriptions.is_fetchable(tp)):
                log.debug('Ignoring    fetched    records    for    partition    %s    since    it    is    no    longer    fetchable', tp)
            elif (error_type is Errors.NoError):
                self._subscriptions.assignment[tp].highwater = highwater
                fetch_offset = fetch_offsets[tp]
                position = self._subscriptions.assignment[tp].position
                if ((position is None) or (position != fetch_offset)):
                    log.debug('Discarding    fetch    response    for    partition    %s    since    its    offset    %d    does    not    match    the    expected    offset    %d', tp, fetch_offset, position)
                    continue
                num_bytes = 0
                partial = None
                if (messages and isinstance(messages[(-1)][(-1)], PartialMessage)):
                    partial = messages.pop()
                if messages:
                    log.debug('Adding    fetched    record    for    partition    %s    with    offset    %d    to    buffered    record    list', tp, position)
                    unpacked = list(self._unpack_message_set(tp, messages))
                    self._records.append(self.PartitionRecords(fetch_offset, tp, unpacked))
                    (last_offset, _, _) = messages[(-1)]
                    self._sensors.records_fetch_lag.record((highwater - last_offset))
                    num_bytes = sum((msg[1] for msg in messages))
                elif partial:
                    self._record_too_large_partitions[tp] = fetch_offset
                self._sensors.record_topic_fetch_metrics(topic, num_bytes, len(messages))
                total_bytes += num_bytes
                total_count += len(messages)
            elif (error_type in (Errors.NotLeaderForPartitionError, Errors.UnknownTopicOrPartitionError)):
                self._client.cluster.request_update()
            elif (error_type is Errors.OffsetOutOfRangeError):
                fetch_offset = fetch_offsets[tp]
                log.info('Fetch    offset    %s    is    out    of    range    for    topic-partition    %s', fetch_offset, tp)
                if self._subscriptions.has_default_offset_reset_policy():
                    self._subscriptions.need_offset_reset(tp)
                    log.info('Resetting    offset    for    topic-partition    %s', tp)
                else:
                    self._offset_out_of_range_partitions[tp] = fetch_offset
            elif (error_type is Errors.TopicAuthorizationFailedError):
                log.warn('Not    authorized    to    read    from    topic    %s.', tp.topic)
                self._unauthorized_topics.add(tp.topic)
            elif (error_type is Errors.UnknownError):
                log.warn('Unknown    error    fetching    data    for    topic-partition    %s', tp)
            else:
                raise error_type('Unexpected    error    while    fetching    data')
    self._sensors.bytes_fetched.record(total_bytes)
    self._sensors.records_fetched.record(total_count)
    if (response.API_VERSION >= 1):
        self._sensors.fetch_throttle_time_sensor.record(response.throttle_time_ms)
    self._sensors.fetch_latency.record(((recv_time - send_time) * 1000))
