@Override
  public boolean execute(Workflow workflow, Task task, WorkflowExecutor executor) {
    try {
      return endJoinExecute(workflow, task);
    } catch (MaestroInternalError | MaestroNotFoundException e) {
      // if end task failed, it is fatal error, no retry
      task.setStatus(Task.Status.FAILED_WITH_TERMINAL_ERROR);
      task.setReasonForIncompletion(e.getMessage());
      LOG.error(
          "Error executing Maestro end task: {} in workflow: {}",
          task.getTaskId(),
          workflow.getWorkflowId(),
          e);
      return true;
    }
    // Don't catch unexpected exception and MaestroWorkflowExecutor will handle it.
  }
@Test
  public void testExecuteReachLeafStepLimit() {
    StepRuntimeState runtimeState = new StepRuntimeState();
    runtimeState.setStatus(StepInstance.Status.RUNNING);
    WorkflowRollupOverview rollup = new WorkflowRollupOverview();
    rollup.setTotalLeafCount(Constants.TOTAL_LEAF_STEP_COUNT_LIMIT + 1);
    SubworkflowArtifact artifact = new SubworkflowArtifact();
    artifact.setSubworkflowOverview(
        WorkflowRuntimeOverview.of(1, new EnumMap<>(StepInstance.Status.class), rollup));
    WorkflowRuntimeSummary runtimeSummary = new WorkflowRuntimeSummary();
    runtimeSummary.setInstanceStatus(WorkflowInstance.Status.IN_PROGRESS);
    runtimeSummary.setRollupBase(rollup);
    testTask.setOutputData(
        twoItemMap(
            "maestro_step_runtime_summary",
            StepRuntimeSummary.builder()
                .stepId("job1")
                .stepAttemptId(2)
                .stepInstanceUuid("bar")
                .stepName("step1")
                .stepInstanceId(123)
                .type(StepType.SUBWORKFLOW)
                .runtimeState(runtimeState)
                .artifacts(Collections.singletonMap("maestro_subworkflow", artifact))
                .build(),
            "maestro_workflow_runtime_summary",
            runtimeSummary));

    endTask.execute(workflow, testTask, null);

    Assert.assertEquals(
        rollup.getTotalLeafCount() * 2,
        ((WorkflowRuntimeSummary) testTask.getOutputData().get("maestro_workflow_runtime_summary"))
            .getRuntimeOverview()
            .getRollupOverview()
            .getTotalLeafCount());

    verify(publisher, times(1)).publish(any(TerminateInstancesJobEvent.class));
  }