<SYSTEM_TASK:>
Solve the following problem using Python, implementing the functions described below, one line at a time
<END_TASK>
<USER_TASK:>
Description:
 def _handle_send_response(self, result, payloadsByTopicPart, deferredsByTopicPart):
 """Handle the response from our client to our send_produce_request This is a bit complex. Failures can happen in a few ways: 1. The client sent an empty list, False, None or some similar thing as the result, but we were expecting real responses. 2. The client had a failure before it even tried sending any requests to any brokers. a. Kafka error: See if we can retry the whole request b. Non-kafka: Figure it's a programming error, fail all deferreds 3. The client sent all the requests (it's all or none) to the brokers but one or more request failed (timed out before receiving a response, or the brokerclient threw some sort of exception on send In this case, the client throws FailedPayloadsError, and attaches the responses (NOTE: some can have errors!), and the payloads where the send itself failed to the exception. 4. The client sent all the requests, all responses were received, but the Kafka broker indicated an error with servicing the request on some of the responses. """


        def _deliver_result(d_list, result=None):
            """Possibly callback each deferred in a list with single result"""
            for d in d_list:
                if not isinstance(d, Deferred):
                    # nested list...
                    _deliver_result(d, result)
                else:
                    # We check d.called since the request could have been
                    # cancelled while we waited for the response
                    if not d.called:
                        d.callback(result)

        def _do_retry(payloads):
            # We use 'fail_on_error=False' because we want our client to
            # process every response that comes back from the brokers so
            # we can determine which requests were successful, and which
            # failed for retry
            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

        def _cancel_retry(failure, dc):
            # Cancel the retry callLater and pass-thru the failure
            dc.cancel()
            # cancel all the top-level deferreds associated with the request
            _deliver_result(deferredsByTopicPart.values(), failure)
            return failure

        def _check_retry_payloads(failed_payloads_with_errs):
            """Check our retry count and retry after a delay or errback

            If we have more retries to try, create a deferred that will fire
            with the result of delayed retry. If not, errback the remaining
            deferreds with failure

            Params:
            failed_payloads - list of (payload, failure) tuples
            """
            # Do we have retries left?
            if self._req_attempts >= self._max_attempts:
                # No, no retries left, fail each failed_payload with its
                # associated failure
                for p, f in failed_payloads_with_errs:
                    t_and_p = TopicAndPartition(p.topic, p.partition)
                    _deliver_result(deferredsByTopicPart[t_and_p], f)
                return
            # Retries remain!  Schedule one...
            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
            # Cancel the callLater when request is cancelled before it fires
            d.addErrback(_cancel_retry, dc)

            # Reset the topic metadata for all topics which had failed_requests
            # where the failures were of the kind UnknownTopicOrPartitionError
            # or NotLeaderForPartitionError, since those indicate our client's
            # metadata is out of date.
            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

        # The payloads we need to retry, if we still can..
        failed_payloads = []
        # In the case we are sending requests without requiring acks, the
        # brokerclient will immediately callback() the deferred upon send with
        # None. In that case, we just iterate over all the deferreds in
        # deferredsByTopicPart and callback them with None
        # If we are expecting responses/acks, and we get an empty result, we
        # callback with a Failure of NoResponseError
        if not result:
            # Success, but no results, is that what we're expecting?
            if self.req_acks == PRODUCER_ACK_NOT_REQUIRED:
                result = None
            else:
                # We got no result, but we were expecting one? Fail everything!
                result = Failure(NoResponseError())
            _deliver_result(deferredsByTopicPart.values(), result)
            return
        elif isinstance(result, Failure):
            # Failure!  Was it total, or partial?
            if not result.check(FailedPayloadsError):
                # Total failure of some sort!
                # The client was unable to send the request at all. If it's
                # a KafkaError (probably Leader/Partition unavailable), retry
                if result.check(KafkaError):
                    # Yep, a kafak error. Set failed_payloads, and we'll retry
                    # them all below. Set failure for errback to callers if we
                    # are all out of retries
                    failure, result = result, []  # no succesful results, retry
                    failed_payloads = [(p, failure) for p in
                                       payloadsByTopicPart.values()]
                else:
                    # Was the request cancelled?
                    if not result.check(tid_CancelledError):
                        # Uh Oh, programming error? Log it!
                        log.error("Unexpected failure: %r in "
                                  "_handle_send_response", result)
                    # Cancelled, or programming error, we fail the requests
                    _deliver_result(deferredsByTopicPart.values(), result)
                    return
            else:
                # FailedPayloadsError: This means that some/all of the
                # requests to a/some brokerclients failed to send.
                # Pull the successful responses and the failed_payloads off
                # the exception and handle them below. Preserve the
                # FailedPayloadsError as 'failure'
                failure = result
                result = failure.value.args[0]
                failed_payloads = failure.value.args[1]

        # Do we have results? Iterate over them and if the response indicates
        # success, then callback the associated deferred. If the response
        # indicates an error, then setup that request for retry.
        # NOTE: In this case, each failed_payload get it's own error...
        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:
                # Success for this topic/partition
                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))

        # Were there any failed requests to possibly retry?
        if failed_payloads:
            return _check_retry_payloads(failed_payloads)
        return