SQS - Reject on failure (#5843)

* reject on failure * add documentation * test fix * test fix * test fix
celery · Nov 22, 2019 · 4e4d308 · 4e4d308
1 parent c4f2692
commit 4e4d308
Show file tree

Hide file tree

Showing 2 changed files with 11 additions and 2 deletions.
diff --git a/celery/worker/request.py b/celery/worker/request.py
@@ -510,6 +510,10 @@ def on_failure(self, exc_info, send_failed_event=True, return_ok=False):
                 send_failed_event = False
             elif ack:
                 self.acknowledge()
+            else:
+                # supporting the behaviour where a task failed and
+                # need to be removed from prefetched local queue
+                self.reject(requeue=False)
 
         # These are special cases where the process would not have had time
         # to write the result.

diff --git a/t/unit/worker/test_request.py b/t/unit/worker/test_request.py
@@ -669,14 +669,17 @@ def test_on_failure_acks_late(self):
     def test_on_failure_acks_on_failure_or_timeout_disabled_for_task(self):
         job = self.xRequest()
         job.time_start = 1
+        job._on_reject = Mock()
         self.mytask.acks_late = True
         self.mytask.acks_on_failure_or_timeout = False
         try:
             raise KeyError('foo')
         except KeyError:
             exc_info = ExceptionInfo()
             job.on_failure(exc_info)
-        assert job.acknowledged is False
+
+        assert job.acknowledged is True
+        job._on_reject.assert_called_with(req_logger, job.connection_errors, False)
 
     def test_on_failure_acks_on_failure_or_timeout_enabled_for_task(self):
         job = self.xRequest()
@@ -701,7 +704,9 @@ def test_on_failure_acks_on_failure_or_timeout_disabled(self):
         except KeyError:
             exc_info = ExceptionInfo()
             job.on_failure(exc_info)
-        assert job.acknowledged is False
+        assert job.acknowledged is True
+        job._on_reject.assert_called_with(req_logger, job.connection_errors,
+                                          False)
         self.app.conf.acks_on_failure_or_timeout = True
 
     def test_on_failure_acks_on_failure_or_timeout_enabled(self):