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)

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