def _produce_request(self, node_id, acks, timeout, batches):
        """Create a produce request from the given record batches.

        Returns:
            ProduceRequest (version depends on client api_versions)
        """
        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

        version = self._client.api_version(ProduceRequest, max_version=7)
        topic_partition_data = [
            (topic, list(partition_info.items()))
            for topic, partition_info in six.iteritems(produce_records_by_partition)]
        transactional_id = self._transaction_manager.transactional_id if self._transaction_manager else None
        if version >= 3:
            return ProduceRequest[version](
                transactional_id=transactional_id,
                required_acks=acks,
                timeout=timeout,
                topics=topic_partition_data,
            )
        else:
            if transactional_id is not None:
                log.warning('%s: Broker does not support ProduceRequest v3+, required for transactional_id', str(self))
            return ProduceRequest[version](
                required_acks=acks,
                timeout=timeout,
                topics=topic_partition_data,
            )
Create a produce request from the given record batches.

        Returns:
            ProduceRequest (version depends on client api_versions)
        
kafka/producer/sender.py