Merge pull request celery#2226 from aneilbaboo/master

Fix issue celery#2225

celery/app/amqp.py

@@ -371,6 +371,7 @@ def as_task_v1(self, task_id, name, args=None, kwargs=None,
                 'id': task_id,
                 'args': args,
                 'kwargs': kwargs,
+                'group': group_id,
                 'retries': retries,
                 'eta': eta,
                 'expires': expires,

celery/backends/base.py

@@ -357,8 +357,8 @@ def fallback_chord_unlock(self, group_id, body, result=None,
 
     def apply_chord(self, header, partial_args, group_id, body,
                     options={}, **kwargs):
-        options['task_id'] = group_id
-        result = header(*partial_args, **options or {})
+        fixed_options = dict((k,v) for k,v in options.items() if k!='task_id')
+        result = header(*partial_args, task_id=group_id, **fixed_options or {})
         self.fallback_chord_unlock(group_id, body, **kwargs)
         return result
 
@@ -534,7 +534,11 @@ def _restore_group(self, group_id):
     def _apply_chord_incr(self, header, partial_args, group_id, body,
                           result=None, options={}, **kwargs):
         self.save_group(group_id, self.app.GroupResult(group_id, result))
-        return header(*partial_args, task_id=group_id, **options or {})
+
+        fixed_options = dict((k,v) for k,v in options.items() if k != 'task_id')
+
+        return header(*partial_args, task_id=group_id, **fixed_options or {})
+
 
     def on_chord_part_return(self, task, state, result, propagate=None):
         if not self.implements_incr:

celery/canvas.py

@@ -474,9 +474,9 @@ def prepare_steps(self, args, tasks,
             if link_error:
                 task.set(link_error=link_error)
 
-            if not isinstance(prev_task, chord):
-                results.append(res)
-                tasks.append(task)
+            tasks.append(task)
+            results.append(res)
+
             prev_task, prev_res = task, res
 
         return tasks, results
@@ -603,7 +603,7 @@ def _maybe_group(tasks):
     elif isinstance(tasks, Signature):
         tasks = [tasks]
     else:
-        tasks = regen(tasks)
+        tasks = map(signature, regen(tasks))
     return tasks