Fix error callback processing for task based class (Fixes celery#4377) (

celery#5232)

* Fix error callback processing for task based class [CELERY-4377]

* Update contributors list.

* Fix linter.

CONTRIBUTORS.txt

@@ -265,3 +265,4 @@ Tom Booth, 2018/07/06
 Axel haustant, 2018/08/14
 Bruno Alla, 2018/09/27
 Artem Vasilyev, 2018/11/24
+Victor Mireyev, 2018/12/13

celery/backends/base.py

@@ -170,6 +170,13 @@ def _call_task_errbacks(self, request, exc, traceback):
         for errback in request.errbacks:
             errback = self.app.signature(errback)
             if (
+                # Celery tasks type created with the @task decorator have the
+                # __header__ property, but Celery task created from Task
+                # class do not have this property.
+                # That's why we have to check if this property exists before
+                # checking is it partial function.
+                hasattr(errback.type, '__header__') and
+
                 # workaround to support tasks with bind=True executed as
                 # link errors. Otherwise retries can't be used
                 not isinstance(errback.type.__header__, partial) and

t/unit/backends/test_base.py

@@ -8,7 +8,7 @@
 from case import ANY, Mock, call, patch, skip
 
 from celery import chord, group, states, uuid
-from celery.app.task import Context
+from celery.app.task import Context, Task
 from celery.backends.base import (BaseBackend, DisabledBackend,
                                   KeyValueStoreBackend, _nulldict)
 from celery.exceptions import ChordError, TimeoutError
@@ -383,6 +383,23 @@ def test_mark_as_failure__errback(self):
         b.mark_as_failure('id', exc, request=request)
         assert self.errback.last_result == 5
 
+    @patch('celery.backends.base.group')
+    def test_class_based_task_can_be_used_as_error_callback(self, mock_group):
+        b = BaseBackend(app=self.app)
+        b._store_result = Mock()
+
+        class TaskBasedClass(Task):
+            def run(self):
+                pass
+
+        TaskBasedClass = self.app.register_task(TaskBasedClass())
+
+        request = Mock(name='request')
+        request.errbacks = [TaskBasedClass.subtask(args=[], immutable=True)]
+        exc = KeyError()
+        b.mark_as_failure('id', exc, request=request)
+        mock_group.assert_called_once_with(request.errbacks, app=self.app)
+
     def test_mark_as_failure__chord(self):
         b = BaseBackend(app=self.app)
         b._store_result = Mock()