anchor.diff:                  true -> finish_pipeline(Stage, S);
                   false -> S
               end;
  -        {Next, Grouping, _} ->
  +        {Next, Grouping, false} ->
  +            start_next_stage(stage_outputs(Stage, S), Next, Grouping, S);
  +        {Next, Grouping, true} ->
               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
  +            case pipeline_utils:group_outputs(Grouping, stage_outputs(Stage, S1)) of
  +                [] -> start_next_stage([], Next, Grouping, S1);
  +                _  -> S1
               end
       end.
   
  
positive.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)};
          {_, Grouping, _} ->
              case jc_utils:stage_info_opt(Stage, SI) of
                  none ->
                      PrevStageOutputs = [{TaskId, Outputs}],
  
negative.code:             end
          end, S, Pending).
  
  -spec send_outputs_to_consumers(state(), [grouped_output()], stage_name()) -> state().
  send_outputs_to_consumers(#state{stage_info = SI} = S, ModifiedGroups, Stage) ->
      #stage_info{group_map = GroupMap} = jc_utils:stage_info(Stage, SI),
  