def _send_requests(self, parts_results, requests):
        """Send the requests

        We've determined the partition for each message group in the batch, or
        got errors for them.
        """
        # We use these dictionaries to be able to combine all the messages
        # destined to the same topic/partition into one request
        # the messages & deferreds, both by topic+partition
        reqsByTopicPart = defaultdict(list)
        payloadsByTopicPart = defaultdict(list)
        deferredsByTopicPart = defaultdict(list)

        # We now have a list of (succeeded/failed, partition/None) tuples
        # for the partition lookups we did on each message group, zipped with
        # the requests
        for (success, part_or_failure), req in zip(parts_results, requests):
            if req.deferred.called:
                # Submitter cancelled the request while we were waiting for
                # the topic/partition, skip it
                continue
            if not success:
                # We failed to get a partition for this request, errback to the
                # caller with the failure. Maybe this should retry? However,
                # since this failure is likely to affect an entire Topic, there
                # should be no issues with ordering of messages within a
                # partition of a topic getting out of order. Let the caller
                # retry the particular request if they like, or they could
                # cancel all their outstanding requests in
                req.deferred.errback(part_or_failure)
                continue
            # Ok, we now have a partition for this request, we can add the
            # request for this topic/partition to reqsByTopicPart, and the
            # caller's deferred to deferredsByTopicPart
            topicPart = TopicAndPartition(req.topic, part_or_failure)
            reqsByTopicPart[topicPart].append(req)
            deferredsByTopicPart[topicPart].append(req.deferred)

        # Build list of payloads grouped by topic/partition
        # That is, we bundle all the messages destined for a given
        # topic/partition, even if they were submitted by different
        # requests into a single 'payload', and then we submit all the
        # payloads as a list to the client for sending to the various
        # brokers. The finest granularity of success/failure is at the
        # payload (topic/partition) level.
        payloads = []
        for (topic, partition), reqs in reqsByTopicPart.items():
            msgSet = create_message_set(reqs, self.codec)
            req = ProduceRequest(topic, partition, msgSet)
            topicPart = TopicAndPartition(topic, partition)
            payloads.append(req)
            payloadsByTopicPart[topicPart] = req
        # Make sure we have some payloads to send
        if not payloads:
            return
        # send the request
        d = self.client.send_produce_request(
            payloads, acks=self.req_acks, timeout=self.ack_timeout,
            fail_on_error=False)
        self._req_attempts += 1
        # add our handlers
        d.addBoth(self._handle_send_response, payloadsByTopicPart,
                  deferredsByTopicPart)
        return d
Send the requests

        We've determined the partition for each message group in the batch, or
        got errors for them.