Task: Retry/signature_from_request should include headers. Closes #2706

celery/app/task.py

@@ -520,9 +520,10 @@ class Task(object):
             'soft_time_limit': limit_soft,
             'time_limit': limit_hard,
             'reply_to': request.reply_to,
+            'headers': request.headers,
         }
         options.update(
-            {'queue': queue} if queue else (request.delivery_info or {})
+            {'queue': queue} if queue else (request.delivery_info or {}),
         )
         return self.signature(
             args, kwargs, options, type=self, **extra_options

celery/tests/tasks/test_tasks.py

@@ -124,6 +124,22 @@ class test_task_retries(TasksCase):
         self.retry_task_noargs.apply(propagate=True).get()
         self.assertEqual(self.retry_task_noargs.iterations, 4)
 
+    def test_signature_from_request__passes_headers(self):
+        self.retry_task.push_request()
+        self.retry_task.request.headers = {'custom': 10.1}
+        sig = self.retry_task.signature_from_request()
+        self.assertEqual(sig.options['headers']['custom'], 10.1)
+
+    def test_signature_from_request__delivery_info(self):
+        self.retry_task.push_request()
+        self.retry_task.request.delivery_info = {
+            'exchange': 'testex',
+            'routing_key': 'testrk',
+        }
+        sig = self.retry_task.signature_from_request()
+        self.assertEqual(sig.options['exchange'], 'testex')
+        self.assertEqual(sig.options['routing_key'], 'testrk')
+
     def test_retry_kwargs_can_be_empty(self):
         self.retry_task_mockapply.push_request()
         try: