Task: Handle retry when task sent to anon-exchange. Closes #3622

celery/app/task.py

@@ -563,9 +563,17 @@ class Task(object):
         args = request.args if args is None else args
         kwargs = request.kwargs if kwargs is None else kwargs
         options = request.as_execution_options()
-        options.update(
-            {'queue': queue} if queue else (request.delivery_info or {}),
-        )
+        if queue:
+            options['queue'] = queue
+        else:
+            delivery_info = request.delivery_info or {}
+            exchange = delivery_info.get('exchange')
+            routing_key = delivery_info.get('routing_key')
+            if exchange == '' and routing_key:
+                # sent to anon-exchange
+                options['queue'] = routing_key
+            else:
+                options.update(delivery_info)
         return self.signature(
             args, kwargs, options, type=self, **extra_options
         )