code: def _handle_fetch_response(self, request, send_time, response):
          
          fetch_offsets = {}
          for topic, partitions in request.topics:
              for partition_data in partitions:
                  partition, offset = partition_data[:2]
                  fetch_offsets[TopicPartition(topic, partition)] = offset
  
          partitions = set([TopicPartition(topic, partition_data[0])
                            for topic, partitions in response.topics
                            for partition_data in partitions])
          metric_aggregator = FetchResponseMetricAggregator(self._sensors, partitions)
  
          # randomized ordering should improve balance for short-lived consumers
          random.shuffle(response.topics)
          for topic, partitions in response.topics:
              random.shuffle(partitions)
              for partition_data in partitions:
                  tp = TopicPartition(topic, partition_data[0])
                  completed_fetch = CompletedFetch(
                      tp, fetch_offsets[tp],
                      response.API_VERSION,
                      partition_data[1:],
                      metric_aggregator
                  )
                  self._completed_fetches.append(completed_fetch)
  
          if response.API_VERSION >= 1:
              self._sensors.fetch_throttle_time_sensor.record(response.throttle_time_ms)
          self._sensors.fetch_latency.record((time.time() - send_time) * 1000)
docs: The callback for fetch completion