Tasks should not be retried if task called directly (__call__). Thanks to subsume

celery/app/task/__init__.py

@@ -29,6 +29,7 @@ class Context(threading.local):
     delivery_info = None
     taskset = None
     chord = None
+    called_directly = True
 
     def update(self, d, **kwargs):
         self.__dict__.update(d, **kwargs)
@@ -496,6 +497,11 @@ class BaseTask(object):
         kwargs = request.kwargs if kwargs is None else kwargs
         delivery_info = request.delivery_info
 
+        # Not in worker or emulated by (apply/always_eager),
+        # so just raise the original exception.
+        if request.called_directly:
+            raise exc or RetryTaskError("Task can be retried", None)
+
         if delivery_info:
             options.setdefault("exchange", delivery_info.get("exchange"))
             options.setdefault("routing_key", delivery_info.get("routing_key"))

celery/execute/trace.py

@@ -72,7 +72,7 @@ class TaskTrace(object):
 
     def execute(self):
         self.task.request.update(self.request, args=self.args,
-                                               kwargs=self.kwargs)
+                                 called_directly=False, kwargs=self.kwargs)
         signals.task_prerun.send(sender=self.task, task_id=self.task_id,
                                  task=self.task, args=self.args,
                                  kwargs=self.kwargs)

celery/tests/test_task/test_task.py

@@ -155,6 +155,7 @@ class TestTaskRetries(unittest.TestCase):
                             args=[4, 4], kwargs=None)
 
     def test_retry_not_eager(self):
+        RetryTaskMockApply.request.called_directly = False
         exc = Exception("baz")
         try:
             RetryTaskMockApply.retry(args=[4, 4], kwargs={"task_retries": 0},