code: 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