anchor.diff:                                                            TaskId, Outputs)
               end,
               S1 = S#state{stage_info = jc_utils:update_stage_tasks(Stage, TaskId, done, SI)},
  -            maybe_submit_tasks(S1, Stage, NewGroups, ModifiedGroups)
  +            {NTasks, S2} = make_stage_tasks(Next, Grouping, NewGroups, S1, {0, []}),
  +            S3 = do_submit_tasks(first_run, NTasks, S2),
  +            send_outputs_to_consumers(S3, ModifiedGroups, Next)
       end.
   
   wakeup_waiters(TaskId, Host, Outputs, #state{tasks = Tasks} = S) ->
  
positive.code:                 true -> finish_pipeline(Stage, S);
                  false -> S
              end;
          {Next, Grouping, _} ->
              S1 = send_termination_signal(Next, S),
              case Stage of
                  ?INPUT ->
                      start_next_stage(stage_outputs(Stage, S1), Next, Grouping, S1);
                  _      ->
                      case pipeline_utils:group_outputs(Grouping,
                                                        stage_outputs(Stage, S1)) of
                          [] -> start_next_stage([], Next, Grouping, S1);
                          _  -> S1
                      end
              end
      end.
  
  
negative.code:         done ->
              % there is no tasks in the next stage to be started.
              S#state{stage_info = jc_utils:update_stage_tasks(Stage, TaskId, done, SI)};
          {_, _, false} ->
              S#state{stage_info = jc_utils:update_stage_tasks(Stage, TaskId, done, SI)};
          {Next, Grouping, true} ->
              case jc_utils:stage_info_opt(Stage, SI) of
                  none ->
                      PrevStageOutputs = [{TaskId, Outputs}],
  