docstr: Create fetch requests for all assigned partitions, grouped by node.
  
  FetchRequests skipped if no leader, or node has requests in flight
  
  Returns:
      dict: {node_id: FetchRequest, ...} (version depends on api_version)
code: def _create_fetch_requests(self):
      """Create fetch requests for all assigned partitions, grouped by node.
  
      FetchRequests skipped if no leader, or node has requests in flight
  
      Returns:
          dict: {node_id: FetchRequest, ...} (version depends on api_version)
      """
      # create the fetch info as a dict of lists of partition info tuples
      # which can be passed to FetchRequest() via .items()
      fetchable = collections.defaultdict(
          lambda: collections.defaultdict(list)
      )
  
      for partition in self._fetchable_partitions():
          node_id = self._client.cluster.leader_for_partition(partition)
  
          # advance position for any deleted compacted messages if required
          if self._subscriptions.assignment[
              partition
          ].last_offset_from_message_batch:
              next_offset_from_batch_header = (
                  self._subscriptions.assignment[
                      partition
                  ].last_offset_from_message_batch
                  + 1
              )
              if (
                  next_offset_from_batch_header
                  > self._subscriptions.assignment[partition].position
              ):
                  log.debug(
                      "Advance position for partition %s from %s to %s (last message batch location plus one)"
                      " to correct for deleted compacted messages",
                      partition,
                      self._subscriptions.assignment[
                          partition
                      ].position,
                      next_offset_from_batch_header,
                  )
                  self._subscriptions.assignment[
                      partition
                  ].position = next_offset_from_batch_header
  
          position = self._subscriptions.assignment[partition].position
  
          # fetch if there is a leader and no in-flight requests
          if node_id is None or node_id == -1:
              log.debug(
                  "No leader found for partition %s."
                  " Requesting metadata update",
                  partition,
              )
              self._client.cluster.request_update()
  
          elif self._client.in_flight_request_count(node_id) == 0:
              partition_info = (
                  partition.partition,
                  position,
                  self.config["max_partition_fetch_bytes"],
              )
              fetchable[node_id][partition.topic].append(partition_info)
              log.debug(
                  "Adding fetch request for partition %s at offset %d",
                  partition,
                  position,
              )
          else:
              log.log(
                  0,
                  "Skipping fetch for partition %s because there is an inflight request to node %s",
                  partition,
                  node_id,
              )
  
      if self.config["api_version"] >= (0, 11, 0):
          version = 4
      elif self.config["api_version"] >= (0, 10, 1):
          version = 3
      elif self.config["api_version"] >= (0, 10):
          version = 2
      elif self.config["api_version"] == (0, 9):
          version = 1
      else:
          version = 0
      requests = {}
      for node_id, partition_data in six.iteritems(fetchable):
          if version < 3:
              requests[node_id] = FetchRequest[version](
                  -1,  # replica_id
                  self.config["fetch_max_wait_ms"],
                  self.config["fetch_min_bytes"],
                  partition_data.items(),
              )
          else:
              # As of version == 3 partitions will be returned in order as
              # they are requested, so to avoid starvation with
              # `fetch_max_bytes` option we need this shuffle
              # NOTE: we do have partition_data in random order due to usage
              #       of unordered structures like dicts, but that does not
              #       guarantee equal distribution, and starting in Python3.6
              #       dicts retain insert order.
              partition_data = list(partition_data.items())
              random.shuffle(partition_data)
              if version == 3:
                  requests[node_id] = FetchRequest[version](
                      -1,  # replica_id
                      self.config["fetch_max_wait_ms"],
                      self.config["fetch_min_bytes"],
                      self.config["fetch_max_bytes"],
                      partition_data,
                  )
              else:
                  requests[node_id] = FetchRequest[version](
                      -1,  # replica_id
                      self.config["fetch_max_wait_ms"],
                      self.config["fetch_min_bytes"],
                      self.config["fetch_max_bytes"],
                      self._isolation_level,
                      partition_data,
                  )
      return requests
  
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 24 for calling the Fetcher obj's underlying member method code with example usage: obj._create_fetch_requests() and returns: requests