Merge pull request #1 from scrapinghub/feature-producer-retries

Async producer: added retries for other error types

Author: vshlapakov

kafka/producer/base.py

@@ -86,30 +86,39 @@ def _send_upstream(queue, client, codec, batch_time, batch_size,
             client.send_produce_request(reqs,
                                         acks=req_acks,
                                         timeout=ack_timeout)
+
         except FailedPayloadsError as ex:
-            failed_reqs = ex.args[0]
-            log.exception("Failed payloads count %s" % len(failed_reqs))
-
-            # if no limit, retry all failed messages until success
-            if retries_limit is None:
-                reqs_to_retry = failed_reqs
-            # makes sense to check failed reqs only if we have a limit > 0
-            elif retries_limit > 0:
-                for req in failed_reqs:
-                    if retries_limit and req.retries < retries_limit:
-                        updated_req = req._replace(retries=req.retries+1)
-                        reqs_to_retry.append(updated_req)
+            log.warning("Async producer send warning: failed payloads.")
+            reqs_to_retry = filter_by_retries(ex.args[0], retries_limit)
+
         except Exception as ex:
-            log.exception("Unable to send message: %s" % type(ex))
-        finally:
-            reqs = []
+            log.error("Async producer send exception: %s" % type(ex))
+            reqs_to_retry = filter_by_retries(reqs, retries_limit)
 
+        reqs = []
         if reqs_to_retry and retry_backoff:
             reqs = reqs_to_retry
             log.warning("%s requests will be retried next call." % len(reqs))
             time.sleep(float(retry_backoff) / 1000)
 
 
+def filter_by_retries(failed_reqs, retries_limit):
+    """ Get requests to retry using retries limit """
+
+    # if no limit, retry all failed messages until success
+    if retries_limit is None:
+        return failed_reqs
+
+    # makes sense to check failed reqs only if we have a limit > 0
+    reqs_to_retry = []
+    if retries_limit > 0:
+        for req in failed_reqs:
+            if req.retries < retries_limit:
+                updated_req = req._replace(retries=req.retries+1)
+                reqs_to_retry.append(updated_req)
+    return reqs_to_retry
+
+
 class Producer(object):
     """
     Base class to be used by producers