code: def _append(self, drained, part, max_records, update_offsets):
      if not part:
          return 0
  
      tp = part.topic_partition
      fetch_offset = part.fetch_offset
      if not self._subscriptions.is_assigned(tp):
          # this can happen when a rebalance happened before
          # fetched records are returned to the consumer's poll call
          log.debug(
              "Not returning fetched records for partition %s"
              " since it is no longer assigned",
              tp,
          )
      else:
          # note that the position should always be available
          # as long as the partition is still assigned
          position = self._subscriptions.assignment[tp].position
          if not self._subscriptions.is_fetchable(tp):
              # this can happen when a partition is paused before
              # fetched records are returned to the consumer's poll call
              log.debug(
                  "Not returning fetched records for assigned partition"
                  " %s since it is no longer fetchable",
                  tp,
              )
  
          elif fetch_offset == position:
              # we are ensured to have at least one record since we already checked for emptiness
              part_records = part.take(max_records)
              next_offset = part_records[-1].offset + 1
  
              log.log(
                  0,
                  "Returning fetched records at offset %d for assigned"
                  " partition %s and update position to %s",
                  position,
                  tp,
                  next_offset,
              )
  
              for record in part_records:
                  drained[tp].append(record)
  
              if update_offsets:
                  self._subscriptions.assignment[
                      tp
                  ].position = next_offset
              return len(part_records)
  
          else:
              # these records aren't next in line based on the last consumed
              # position, ignore them they must be from an obsolete request
              log.debug(
                  "Ignoring fetched records for %s at offset %s since"
                  " the current position is %d",
                  tp,
                  part.fetch_offset,
                  position,
              )
  
      part.discard()
      return 0
  
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 14 for calling the Fetcher obj's underlying member method code with example usage: obj._append(drained, part, max_records, update_offsets) and returns: int, int, len