'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):

    fetchable = collections.defaultdict((lambda : collections.defaultdict(list)))
    for partition in self._fetchable_partitions():
        node_id = self._client.cluster.leader_for_partition(partition)
        position = self._subscriptions.assignment[partition].position
        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)
    if (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), self.config['fetch_max_wait_ms'], self.config['fetch_min_bytes'], partition_data.items())
        else:
            partition_data = list(partition_data.items())
            random.shuffle(partition_data)
            requests[node_id] = FetchRequest[version]((-1), self.config['fetch_max_wait_ms'], self.config['fetch_min_bytes'], self.config['fetch_max_bytes'], partition_data)
    return requests
