def _create_produce_requests(self, collated):
        """
        Transfer the record batches into a list of produce requests on a
        per-node basis.

        Arguments:
            collated: {node_id: [ProducerBatch]}

        Returns:
            dict: {node_id: ProduceRequest} (version depends on client api_versions)
        """
        requests = {}
        for node_id, batches in six.iteritems(collated):
            if batches:
                requests[node_id] = self._produce_request(
                    node_id, self.config['acks'],
                    self.config['request_timeout_ms'], batches)
        return requests

        Transfer the record batches into a list of produce requests on a
        per-node basis.

        Arguments:
            collated: {node_id: [ProducerBatch]}

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