code: def _parse_fetched_data(self, completed_fetch):
      tp = completed_fetch.topic_partition
      fetch_offset = completed_fetch.fetched_offset
      num_bytes = 0
      records_count = 0
      parsed_records = None
  
      error_code, highwater = completed_fetch.partition_data[:2]
      error_type = Errors.for_code(error_code)
  
      try:
          if not self._subscriptions.is_fetchable(tp):
              # this can happen when a rebalance happened or a partition
              # consumption paused while fetch is still in-flight
              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
  
              # we are interested in this fetch only if the beginning
              # offset (of the *request*) matches the current consumed position
              # Note that the *response* may return a messageset that starts
              # earlier (e.g., compressed messages) or later (e.g., compacted topic)
              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,
                  )
                  return None
  
              records = MemoryRecords(
                  completed_fetch.partition_data[-1]
              )
              if records.has_next():
                  log.debug(
                      "Adding fetched record for partition %s with"
                      " offset %d to buffered record list",
                      tp,
                      position,
                  )
                  unpacked = list(self._unpack_message_set(tp, records))
                  parsed_records = self.PartitionRecords(
                      fetch_offset, tp, unpacked
                  )
                  if unpacked:
                      last_offset = unpacked[-1].offset
                      self._sensors.records_fetch_lag.record(
                          highwater - last_offset
                      )
                  num_bytes = records.valid_bytes()
                  records_count = len(unpacked)
              elif records.size_in_bytes() > 0:
                  # we did not read a single message from a non-empty
                  # buffer because that message's size is larger than
                  # fetch size, in this case record this exception
                  record_too_large_partitions = {tp: fetch_offset}
                  raise RecordTooLargeError(
                      "There are some messages at [Partition=Offset]: %s "
                      " whose size is larger than the fetch size %s"
                      " and hence cannot be ever returned."
                      " Increase the fetch size, or decrease the maximum message"
                      " size the broker will allow."
                      % (
                          record_too_large_partitions,
                          self.config["max_partition_fetch_bytes"],
                      ),
                      record_too_large_partitions,
                  )
              self._sensors.record_topic_fetch_metrics(
                  tp.topic, num_bytes, records_count
              )
  
          elif error_type in (
              Errors.NotLeaderForPartitionError,
              Errors.UnknownTopicOrPartitionError,
          ):
              self._client.cluster.request_update()
          elif error_type is Errors.OffsetOutOfRangeError:
              position = self._subscriptions.assignment[tp].position
              if position is None or position != fetch_offset:
                  log.debug(
                      "Discarding stale fetch response for partition %s"
                      " since the fetched offset %d does not match the"
                      " current offset %d",
                      tp,
                      fetch_offset,
                      position,
                  )
              elif (
                  self._subscriptions.has_default_offset_reset_policy()
              ):
                  log.info(
                      "Fetch offset %s is out of range for topic-partition %s",
                      fetch_offset,
                      tp,
                  )
                  self._subscriptions.need_offset_reset(tp)
              else:
                  raise Errors.OffsetOutOfRangeError({tp: fetch_offset})
  
          elif error_type is Errors.TopicAuthorizationFailedError:
              log.warning(
                  "Not authorized to read from topic %s.", tp.topic
              )
              raise Errors.TopicAuthorizationFailedError(set(tp.topic))
          elif error_type is Errors.UnknownError:
              log.warning(
                  "Unknown error fetching data for topic-partition %s",
                  tp,
              )
          else:
              raise error_type("Unexpected error while fetching data")
  
      finally:
          completed_fetch.metric_aggregator.record(
              tp, num_bytes, records_count
          )
  
      return parsed_records
  
imports: from __future__ import absolute_import
  import collections
  import copy
  import logging
  import random
  import sys
  import time
  from kafka.vendor import six
  import kafka.errors
  from kafka.future import Future
  from kafka.metrics.stats import Avg, Count, Max, Rate
  from kafka.protocol.fetch import FetchRequest
  from kafka.protocol.offset import OffsetRequest, OffsetResetStrategy, UNKNOWN_OFFSET
  from kafka.record import MemoryRecords
  from kafka.serializer import Deserializer
  from kafka.structs import TopicPartition, OffsetAndTimestamp
label_desc: Use image node_id 26 for calling the Fetcher obj's underlying member method code with example usage: obj._parse_fetched_data(completed_fetch) and returns: parsed_records, None