def _handle_send_response(self, result, payloadsByTopicPart,
                              deferredsByTopicPart):
        def _deliver_result(d_list, result=None):
            for d in d_list:
                if not isinstance(d, Deferred):
                    _deliver_result(d, result)
                else:
                    if not d.called:
                        d.callback(result)
        def _do_retry(payloads):
            d = self.client.send_produce_request(
                payloads, acks=self.req_acks, timeout=self.ack_timeout,
                fail_on_error=False)
            self._req_attempts += 1
            d.addBoth(self._handle_send_response, payloadsByTopicPart,
                      deferredsByTopicPart)
            return d
        def _cancel_retry(failure, dc):
            dc.cancel()
            _deliver_result(deferredsByTopicPart.values(), failure)
            return failure
        def _check_retry_payloads(failed_payloads_with_errs):
            if self._req_attempts >= self._max_attempts:
                for p, f in failed_payloads_with_errs:
                    t_and_p = TopicAndPartition(p.topic, p.partition)
                    _deliver_result(deferredsByTopicPart[t_and_p], f)
                return
            d = Deferred()
            dc = self.client.reactor.callLater(
                self._retry_interval, d.callback, [p for p, f in
                                                   failed_payloads])
            self._retry_interval *= self.RETRY_INTERVAL_FACTOR
            d.addErrback(_cancel_retry, dc)
            reset_topics = set()
            for payload, e in failed_payloads:
                if (isinstance(e, NotLeaderForPartitionError) or
                        isinstance(e, UnknownTopicOrPartitionError)):
                    reset_topics.add(payload.topic)
            if reset_topics:
                self.client.reset_topic_metadata(*reset_topics)
            d.addCallback(_do_retry)
            return d
        failed_payloads = []
        if not result:
            if self.req_acks == PRODUCER_ACK_NOT_REQUIRED:
                result = None
            else:
                result = Failure(NoResponseError())
            _deliver_result(deferredsByTopicPart.values(), result)
            return
        elif isinstance(result, Failure):
            if not result.check(FailedPayloadsError):
                if result.check(KafkaError):
                    failure, result = result, []
                    failed_payloads = [(p, failure) for p in
                                       payloadsByTopicPart.values()]
                else:
                    if not result.check(tid_CancelledError):
                        log.error("Unexpected failure: %r in "
                                  "_handle_send_response", result)
                    _deliver_result(deferredsByTopicPart.values(), result)
                    return
            else:
                failure = result
                result = failure.value.args[0]
                failed_payloads = failure.value.args[1]
        for res in result:
            t_and_p = TopicAndPartition(res.topic, res.partition)
            t_and_p_err = _check_error(res, raiseException=False)
            if not t_and_p_err:
                d_list = deferredsByTopicPart[t_and_p]
                _deliver_result(d_list, res)
            else:
                p = payloadsByTopicPart[t_and_p]
                failed_payloads.append((p, t_and_p_err))
        if failed_payloads:
            return _check_retry_payloads(failed_payloads)
        return