'Create a produce request from the given record batches.
Returns:
ProduceRequest (version depends on api_version)'

def _produce_request(self, node_id, acks, timeout, batches):

    produce_records_by_partition = collections.defaultdict(dict)
    for batch in batches:
        topic = batch.topic_partition.topic
        partition = batch.topic_partition.partition
        buf = batch.records.buffer()
        produce_records_by_partition[topic][partition] = buf
    if (self.config['api_version'] >= (0, 10)):
        version = 2
    elif (self.config['api_version'] == (0, 9)):
        version = 1
    else:
        version = 0
    return ProduceRequest[version](required_acks=acks, timeout=timeout, topics=[(topic, list(partition_info.items())) for (topic, partition_info) in six.iteritems(produce_records_by_partition)])
