Store task args inside MaxRetries exception (celery#5213)

Allow to easily get the parent task's parameters from an error
callback

celery/app/task.py

@@ -689,7 +689,9 @@ def retry(self, args=None, kwargs=None, exc=None, throw=True,
                 raise_with_context(exc)
             raise self.MaxRetriesExceededError(
                 "Can't retry {0}[{1}] args:{2} kwargs:{3}".format(
-                    self.name, request.id, S.args, S.kwargs))
+                    self.name, request.id, S.args, S.kwargs
+                ), task_args=S.args, task_kwargs=S.kwargs
+            )
 
         ret = Retry(exc=exc, when=eta or countdown)
 

celery/exceptions.py

@@ -223,6 +223,11 @@ class TimeoutError(TaskError):
 class MaxRetriesExceededError(TaskError):
     """The tasks max restart limit has been exceeded."""
 
+    def __init__(self, *args, **kwargs):
+        self.task_args = kwargs.pop("task_args", [])
+        self.task_kwargs = kwargs.pop("task_kwargs", dict())
+        super(MaxRetriesExceededError, self).__init__(*args, **kwargs)
+
 
 class TaskRevokedError(TaskError):
     """The task has been revoked, so no result available."""

t/unit/tasks/test_tasks.py

@@ -273,6 +273,18 @@ def test_max_retries_exceeded(self):
             result.get()
         assert self.retry_task.iterations == 2
 
+    def test_max_retries_exceeded_task_args(self):
+        self.retry_task.max_retries = 2
+        self.retry_task.iterations = 0
+        args = (0xFF, 0xFFFF)
+        kwargs = {'care': False}
+        result = self.retry_task.apply(args, kwargs)
+        with pytest.raises(self.retry_task.MaxRetriesExceededError) as e:
+            result.get()
+
+        assert e.value.task_args == args
+        assert e.value.task_kwargs == kwargs
+
     def test_autoretry_no_kwargs(self):
         self.autoretry_task_no_kwargs.max_retries = 3
         self.autoretry_task_no_kwargs.iterations = 0