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, {TopicPartition: fetch_offset}), ...} (version depends on client api_versions)
        """
        # create the fetch info as a dict of lists of partition info tuples
        # which can be passed to FetchRequest() via .items()
        version = self._client.api_version(FetchRequest, max_version=10)
        fetchable = collections.defaultdict(collections.OrderedDict)

        for partition in self._fetchable_partitions():
            node_id = self._client.cluster.leader_for_partition(partition)

            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 not self._client.connected(node_id) and self._client.connection_delay(node_id) > 0:
                # If we try to send during the reconnect backoff window, then the request is just
                # going to be failed anyway before being sent, so skip the send for now
                log.debug("Skipping fetch for partition %s because node %s is awaiting reconnect backoff",
                        partition, node_id)

            elif self._client.throttle_delay(node_id) > 0:
                # If we try to send while throttled, then the request is just
                # going to be failed anyway before being sent, so skip the send for now
                log.debug("Skipping fetch for partition %s because node %s is throttled",
                        partition, node_id)

            elif not self._client.ready(node_id):
                # Until we support send request queues, any attempt to send to a not-ready node will be
                # immediately failed with NodeNotReadyError.
                log.debug("Skipping fetch for partition %s because connection to leader node is not ready yet")

            elif node_id in self._nodes_with_pending_fetch_requests:
                log.debug("Skipping fetch for partition %s because there is a pending fetch request to node %s",
                        partition, node_id)

            else:
                # Leader is connected and does not have a pending fetch request
                if version < 5:
                    partition_info = (
                        partition.partition,
                        position.offset,
                        self.config['max_partition_fetch_bytes']
                    )
                elif version <= 8:
                    partition_info = (
                        partition.partition,
                        position.offset,
                        -1, # log_start_offset is used internally by brokers / replicas only
                        self.config['max_partition_fetch_bytes'],
                    )
                else:
                    partition_info = (
                        partition.partition,
                        position.leader_epoch,
                        position.offset,
                        -1, # log_start_offset is used internally by brokers / replicas only
                        self.config['max_partition_fetch_bytes'],
                    )

                fetchable[node_id][partition] = partition_info
                log.debug("Adding fetch request for partition %s at offset %d",
                          partition, position.offset)

        requests = {}
        for node_id, next_partitions in six.iteritems(fetchable):
            if version >= 7 and self.config['enable_incremental_fetch_sessions']:
                if node_id not in self._session_handlers:
                    self._session_handlers[node_id] = FetchSessionHandler(node_id)
                session = self._session_handlers[node_id].build_next(next_partitions)
            else:
                # No incremental fetch support
                session = FetchRequestData(next_partitions, None, FetchMetadata.LEGACY)

            if version <= 2:
                request = FetchRequest[version](
                    -1,  # replica_id
                    self.config['fetch_max_wait_ms'],
                    self.config['fetch_min_bytes'],
                    session.to_send)
            elif version == 3:
                request = FetchRequest[version](
                    -1,  # replica_id
                    self.config['fetch_max_wait_ms'],
                    self.config['fetch_min_bytes'],
                    self.config['fetch_max_bytes'],
                    session.to_send)
            elif version <= 6:
                request = FetchRequest[version](
                    -1,  # replica_id
                    self.config['fetch_max_wait_ms'],
                    self.config['fetch_min_bytes'],
                    self.config['fetch_max_bytes'],
                    self._isolation_level,
                    session.to_send)
            else:
                # Through v8
                request = FetchRequest[version](
                    -1,  # replica_id
                    self.config['fetch_max_wait_ms'],
                    self.config['fetch_min_bytes'],
                    self.config['fetch_max_bytes'],
                    self._isolation_level,
                    session.id,
                    session.epoch,
                    session.to_send,
                    session.to_forget)

            fetch_offsets = {}
            for tp, partition_data in six.iteritems(next_partitions):
                if version <= 8:
                    offset = partition_data[1]
                else:
                    offset = partition_data[2]
                fetch_offsets[tp] = offset

            requests[node_id] = (request, fetch_offsets)

        return requests
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, {TopicPartition: fetch_offset}), ...} (version depends on client api_versions)
        
kafka/consumer/fetcher.py