nodes: 0, module; 1, function_definition; 2, function_name:_handle_send_response; 3, parameters; 4, block; 5, identifier:self; 6, identifier:result; 7, identifier:payloadsByTopicPart; 8, identifier:deferredsByTopicPart; 9, expression_statement; 10, function_definition; 11, function_definition; 12, function_definition; 13, function_definition; 14, comment:# The payloads we need to retry, if we still can..; 15, expression_statement; 16, comment:# In the case we are sending requests without requiring acks, the; 17, comment:# brokerclient will immediately callback() the deferred upon send with; 18, comment:# None. In that case, we just iterate over all the deferreds in; 19, comment:# deferredsByTopicPart and callback them with None; 20, comment:# If we are expecting responses/acks, and we get an empty result, we; 21, comment:# callback with a Failure of NoResponseError; 22, if_statement; 23, comment:# Do we have results? Iterate over them and if the response indicates; 24, comment:# success, then callback the associated deferred. If the response; 25, comment:# indicates an error, then setup that request for retry.; 26, comment:# NOTE: In this case, each failed_payload get it's own error...; 27, for_statement; 28, comment:# Were there any failed requests to possibly retry?; 29, if_statement; 30, return_statement; 31, comment:"""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.
          """; 32, function_name:_deliver_result; 33, parameters; 34, block; 35, function_name:_do_retry; 36, parameters; 37, comment:# We use 'fail_on_error=False' because we want our client to; 38, comment:# process every response that comes back from the brokers so; 39, comment:# we can determine which requests were successful, and which; 40, comment:# failed for retry; 41, block; 42, function_name:_cancel_retry; 43, parameters; 44, comment:# Cancel the retry callLater and pass-thru the failure; 45, block; 46, function_name:_check_retry_payloads; 47, parameters; 48, block; 49, assignment; 50, not_operator; 51, comment:# Success, but no results, is that what we're expecting?; 52, block; 53, elif_clause; 54, identifier:res; 55, identifier:result; 56, block; 57, identifier:failed_payloads; 58, block; 59, identifier:d_list; 60, default_parameter; 61, expression_statement; 62, for_statement; 63, identifier:payloads; 64, expression_statement; 65, expression_statement; 66, comment:# add our handlers; 67, expression_statement; 68, return_statement; 69, identifier:failure; 70, identifier:dc; 71, expression_statement; 72, comment:# cancel all the top-level deferreds associated with the request; 73, expression_statement; 74, return_statement; 75, identifier:failed_payloads_with_errs; 76, expression_statement; 77, comment:# Do we have retries left?; 78, if_statement; 79, comment:# Retries remain!  Schedule one...; 80, expression_statement; 81, expression_statement; 82, expression_statement; 83, comment:# Cancel the callLater when request is cancelled before it fires; 84, expression_statement; 85, comment:# Reset the topic metadata for all topics which had failed_requests; 86, comment:# where the failures were of the kind UnknownTopicOrPartitionError; 87, comment:# or NotLeaderForPartitionError, since those indicate our client's; 88, comment:# metadata is out of date.; 89, expression_statement; 90, for_statement; 91, if_statement; 92, expression_statement; 93, return_statement; 94, identifier:failed_payloads; 95, list; 96, identifier:result; 97, if_statement; 98, expression_statement; 99, return_statement; 100, call; 101, comment:# Failure!  Was it total, or partial?; 102, block; 103, expression_statement; 104, expression_statement; 105, if_statement; 106, return_statement; 107, identifier:result; 108, None; 109, comment:"""Possibly callback each deferred in a list with single result"""; 110, identifier:d; 111, identifier:d_list; 112, block; 113, assignment; 114, augmented_assignment; 115, call; 116, identifier:d; 117, call; 118, call; 119, identifier:failure; 120, comment:"""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
              """; 121, comparison_operator:self._req_attempts >= self._max_attempts; 122, comment:# No, no retries left, fail each failed_payload with its; 123, comment:# associated failure; 124, block; 125, assignment; 126, assignment; 127, augmented_assignment; 128, call; 129, assignment; 130, pattern_list; 131, identifier:failed_payloads; 132, block; 133, identifier:reset_topics; 134, block; 135, call; 136, identifier:d; 137, comparison_operator:self.req_acks == PRODUCER_ACK_NOT_REQUIRED; 138, block; 139, else_clause; 140, call; 141, identifier:isinstance; 142, argument_list; 143, if_statement; 144, assignment; 145, assignment; 146, not_operator; 147, comment:# Success for this topic/partition; 148, block; 149, else_clause; 150, call; 151, if_statement; 152, identifier:d; 153, call; 154, attribute; 155, integer:1; 156, attribute; 157, argument_list; 158, attribute; 159, argument_list; 160, identifier:_deliver_result; 161, argument_list; 162, attribute; 163, attribute; 164, for_statement; 165, return_statement; 166, identifier:d; 167, call; 168, identifier:dc; 169, call; 170, attribute; 171, attribute; 172, attribute; 173, argument_list; 174, identifier:reset_topics; 175, call; 176, identifier:payload; 177, identifier:e; 178, if_statement; 179, expression_statement; 180, attribute; 181, argument_list; 182, attribute; 183, identifier:PRODUCER_ACK_NOT_REQUIRED; 184, expression_statement; 185, comment:# We got no result, but we were expecting one? Fail everything!; 186, block; 187, identifier:_deliver_result; 188, argument_list; 189, identifier:result; 190, identifier:Failure; 191, not_operator; 192, comment:# Total failure of some sort!; 193, comment:# The client was unable to send the request at all. If it's; 194, comment:# a KafkaError (probably Leader/Partition unavailable), retry; 195, block; 196, else_clause; 197, identifier:t_and_p; 198, call; 199, identifier:t_and_p_err; 200, call; 201, identifier:t_and_p_err; 202, expression_statement; 203, expression_statement; 204, block; 205, identifier:_check_retry_payloads; 206, argument_list; 207, not_operator; 208, comment:# nested list...; 209, block; 210, else_clause; 211, attribute; 212, argument_list; 213, identifier:self; 214, identifier:_req_attempts; 215, identifier:d; 216, identifier:addBoth; 217, attribute; 218, identifier:payloadsByTopicPart; 219, identifier:deferredsByTopicPart; 220, identifier:dc; 221, identifier:cancel; 222, call; 223, identifier:failure; 224, identifier:self; 225, identifier:_req_attempts; 226, identifier:self; 227, identifier:_max_attempts; 228, pattern_list; 229, identifier:failed_payloads_with_errs; 230, block; 231, identifier:Deferred; 232, argument_list; 233, attribute; 234, argument_list; 235, identifier:self; 236, identifier:_retry_interval; 237, identifier:self; 238, identifier:RETRY_INTERVAL_FACTOR; 239, identifier:d; 240, identifier:addErrback; 241, identifier:_cancel_retry; 242, identifier:dc; 243, identifier:set; 244, argument_list; 245, parenthesized_expression; 246, block; 247, call; 248, identifier:d; 249, identifier:addCallback; 250, identifier:_do_retry; 251, identifier:self; 252, identifier:req_acks; 253, assignment; 254, expression_statement; 255, call; 256, identifier:result; 257, call; 258, if_statement; 259, comment:# FailedPayloadsError: This means that some/all of the; 260, comment:# requests to a/some brokerclients failed to send.; 261, comment:# Pull the successful responses and the failed_payloads off; 262, comment:# the exception and handle them below. Preserve the; 263, comment:# FailedPayloadsError as 'failure'; 264, block; 265, identifier:TopicAndPartition; 266, argument_list; 267, identifier:_check_error; 268, argument_list; 269, assignment; 270, call; 271, expression_statement; 272, expression_statement; 273, identifier:failed_payloads; 274, call; 275, expression_statement; 276, comment:# We check d.called since the request could have been; 277, comment:# cancelled while we waited for the response; 278, block; 279, attribute; 280, identifier:send_produce_request; 281, identifier:payloads; 282, keyword_argument; 283, keyword_argument; 284, keyword_argument; 285, identifier:self; 286, identifier:_handle_send_response; 287, attribute; 288, argument_list; 289, identifier:p; 290, identifier:f; 291, expression_statement; 292, expression_statement; 293, attribute; 294, identifier:callLater; 295, attribute; 296, attribute; 297, list_comprehension; 298, boolean_operator; 299, expression_statement; 300, attribute; 301, argument_list; 302, identifier:result; 303, None; 304, assignment; 305, attribute; 306, argument_list; 307, attribute; 308, argument_list; 309, call; 310, comment:# Yep, a kafak error. Set failed_payloads, and we'll retry; 311, comment:# them all below. Set failure for errback to callers if we; 312, comment:# are all out of retries; 313, block; 314, else_clause; 315, expression_statement; 316, expression_statement; 317, expression_statement; 318, attribute; 319, attribute; 320, identifier:res; 321, keyword_argument; 322, identifier:d_list; 323, subscript; 324, identifier:_deliver_result; 325, argument_list; 326, assignment; 327, call; 328, identifier:isinstance; 329, argument_list; 330, call; 331, if_statement; 332, identifier:self; 333, identifier:client; 334, identifier:acks; 335, attribute; 336, identifier:timeout; 337, attribute; 338, identifier:fail_on_error; 339, False; 340, identifier:deferredsByTopicPart; 341, identifier:values; 342, assignment; 343, call; 344, attribute; 345, identifier:reactor; 346, identifier:self; 347, identifier:_retry_interval; 348, identifier:d; 349, identifier:callback; 350, identifier:p; 351, for_in_clause; 352, call; 353, call; 354, call; 355, attribute; 356, identifier:reset_topic_metadata; 357, list_splat; 358, identifier:result; 359, call; 360, identifier:deferredsByTopicPart; 361, identifier:values; 362, identifier:result; 363, identifier:check; 364, identifier:FailedPayloadsError; 365, attribute; 366, argument_list; 367, expression_statement; 368, comment:# no succesful results, retry; 369, expression_statement; 370, comment:# Was the request cancelled?; 371, block; 372, assignment; 373, assignment; 374, assignment; 375, identifier:res; 376, identifier:topic; 377, identifier:res; 378, identifier:partition; 379, identifier:raiseException; 380, False; 381, identifier:deferredsByTopicPart; 382, identifier:t_and_p; 383, identifier:d_list; 384, identifier:res; 385, identifier:p; 386, subscript; 387, attribute; 388, argument_list; 389, identifier:d; 390, identifier:Deferred; 391, identifier:_deliver_result; 392, argument_list; 393, not_operator; 394, block; 395, identifier:self; 396, identifier:req_acks; 397, identifier:self; 398, identifier:ack_timeout; 399, identifier:t_and_p; 400, call; 401, identifier:_deliver_result; 402, argument_list; 403, identifier:self; 404, identifier:client; 405, pattern_list; 406, identifier:failed_payloads; 407, identifier:isinstance; 408, argument_list; 409, identifier:isinstance; 410, argument_list; 411, attribute; 412, argument_list; 413, identifier:self; 414, identifier:client; 415, identifier:reset_topics; 416, identifier:Failure; 417, argument_list; 418, identifier:result; 419, identifier:check; 420, identifier:KafkaError; 421, assignment; 422, assignment; 423, if_statement; 424, comment:# Cancelled, or programming error, we fail the requests; 425, expression_statement; 426, return_statement; 427, identifier:failure; 428, identifier:result; 429, identifier:result; 430, subscript; 431, identifier:failed_payloads; 432, subscript; 433, identifier:payloadsByTopicPart; 434, identifier:t_and_p; 435, identifier:failed_payloads; 436, identifier:append; 437, tuple; 438, identifier:d; 439, identifier:result; 440, attribute; 441, expression_statement; 442, identifier:TopicAndPartition; 443, argument_list; 444, subscript; 445, identifier:f; 446, identifier:p; 447, identifier:f; 448, identifier:e; 449, identifier:NotLeaderForPartitionError; 450, identifier:e; 451, identifier:UnknownTopicOrPartitionError; 452, identifier:reset_topics; 453, identifier:add; 454, attribute; 455, call; 456, pattern_list; 457, expression_list; 458, identifier:failed_payloads; 459, list_comprehension; 460, not_operator; 461, comment:# Uh Oh, programming error? Log it!; 462, block; 463, call; 464, attribute; 465, integer:0; 466, attribute; 467, integer:1; 468, identifier:p; 469, identifier:t_and_p_err; 470, identifier:d; 471, identifier:called; 472, call; 473, attribute; 474, attribute; 475, identifier:deferredsByTopicPart; 476, identifier:t_and_p; 477, identifier:payload; 478, identifier:topic; 479, identifier:NoResponseError; 480, argument_list; 481, identifier:failure; 482, identifier:result; 483, identifier:result; 484, list; 485, tuple; 486, for_in_clause; 487, call; 488, expression_statement; 489, identifier:_deliver_result; 490, argument_list; 491, attribute; 492, identifier:args; 493, attribute; 494, identifier:args; 495, attribute; 496, argument_list; 497, identifier:p; 498, identifier:topic; 499, identifier:p; 500, identifier:partition; 501, identifier:p; 502, identifier:failure; 503, identifier:p; 504, call; 505, attribute; 506, argument_list; 507, call; 508, call; 509, identifier:result; 510, identifier:failure; 511, identifier:value; 512, identifier:failure; 513, identifier:value; 514, identifier:d; 515, identifier:callback; 516, identifier:result; 517, attribute; 518, argument_list; 519, identifier:result; 520, identifier:check; 521, identifier:tid_CancelledError; 522, attribute; 523, argument_list; 524, attribute; 525, argument_list; 526, identifier:payloadsByTopicPart; 527, identifier:values; 528, identifier:log; 529, identifier:error; 530, concatenated_string; 531, identifier:result; 532, identifier:deferredsByTopicPart; 533, identifier:values; 534, string:"Unexpected failure: %r in "; 535, string:"_handle_send_response"
code: 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