docs: 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