Fix max_retries override on self.retry (#6436)

* Fix max_retries override

* Fix max_retries override

* Fix max_retries override

* Update exceptions.py

typo

* Update autoretry.py

typo

* Update task.py

Prevent exception unpacking for tasks without autoretry_for

* Update test_tasks.py

Unit test

* Update test_tasks.py

Added a new test

* Update autoretry.py

Fox for explicit raise in tasks

* Update test_tasks.py

* Update autoretry.py

* Update task.py

* Update exceptions.py

* Update task.py

celery/app/autoretry.py

@@ -46,6 +46,15 @@ def run(*args, **kwargs):
                             retries=task.request.retries,
                             maximum=retry_backoff_max,
                             full_jitter=retry_jitter)
-                raise task.retry(exc=exc, **retry_kwargs)
+                # Override max_retries
+                if hasattr(task, 'override_max_retries'):
+                    retry_kwargs['max_retries'] =  getattr(task,
+                                                      'override_max_retries',
+                                                      task.max_retries)
+                ret = task.retry(exc=exc, **retry_kwargs)
+                # Stop propagation
+                if hasattr(task, 'override_max_retries'):
+                    delattr(task, 'override_max_retries')
+                raise ret
 
         task._orig_run, task.run = task.run, run

celery/app/task.py

@@ -675,6 +675,8 @@ def retry(self, args=None, kwargs=None, exc=None, throw=True,
         """
         request = self.request
         retries = request.retries + 1
+        if max_retries is not None:
+            self.override_max_retries = max_retries
         max_retries = self.max_retries if max_retries is None else max_retries
 
         # Not in worker or emulated by (apply/always_eager),

celery/exceptions.py

@@ -293,3 +293,4 @@ def __init__(self, *args, **kwargs):
 
     def __repr__(self):
         return super().__repr__() + " state:" + self.state + " task_id:" + self.task_id
+

t/unit/tasks/test_tasks.py

@@ -144,6 +144,27 @@ def retry_task_auto_retry_exception_with_new_args(self, ret=None, place_holder=N
 
         self.retry_task_auto_retry_exception_with_new_args = retry_task_auto_retry_exception_with_new_args
 
+        @self.app.task(bind=True, max_retries=10, iterations=0, shared=False,
+                       autoretry_for=(Exception,))
+        def retry_task_max_retries_override(self, **kwargs):
+            # Test for #6436
+            self.iterations += 1
+            if self.iterations == 3:
+                # I wanna force fail here cause i have enough
+                self.retry(exc=MyCustomException, max_retries=0)
+            self.retry(exc=MyCustomException)
+
+        self.retry_task_max_retries_override = retry_task_max_retries_override
+
+        @self.app.task(bind=True, max_retries=0, iterations=0, shared=False,
+                       autoretry_for=(Exception,))
+        def retry_task_explicit_exception(self, **kwargs):
+            # Test for #6436
+            self.iterations += 1
+            raise MyCustomException()
+
+        self.retry_task_explicit_exception = retry_task_explicit_exception
+
         @self.app.task(bind=True, max_retries=3, iterations=0, shared=False)
         def retry_task_raise_without_throw(self, **kwargs):
             self.iterations += 1
@@ -432,6 +453,22 @@ def test_eager_retry_with_new_params(self):
     def test_eager_retry_with_autoretry_for_exception(self):
         assert self.retry_task_auto_retry_exception_with_new_args.si(place_holder="test").apply().get() == "test"
 
+    def test_retry_task_max_retries_override(self):
+        self.retry_task_max_retries_override.max_retries = 10
+        self.retry_task_max_retries_override.iterations = 0
+        result = self.retry_task_max_retries_override.apply()
+        with pytest.raises(MyCustomException):
+            result.get()
+        assert self.retry_task_max_retries_override.iterations == 3
+
+    def test_retry_task_explicit_exception(self):
+        self.retry_task_explicit_exception.max_retries = 0
+        self.retry_task_explicit_exception.iterations = 0
+        result = self.retry_task_explicit_exception.apply()
+        with pytest.raises(MyCustomException):
+            result.get()
+        assert self.retry_task_explicit_exception.iterations == 1
+
     def test_retry_eager_should_return_value(self):
         self.retry_task.max_retries = 3
         self.retry_task.iterations = 0