@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 testMarkMaestroWorkflowStarted() {
    workflow.setEvent("234567");
    testTask.setReferenceTaskName("maestro_start");
    testTask.setStartTime(123456L);
    testTask.setOutputData(
        twoItemMap(
            "maestro_step_runtime_summary",
            StepRuntimeSummary.builder()
                .stepId("job1")
                .stepAttemptId(2)
                .stepInstanceUuid("bar")
                .stepName("step1")
                .stepInstanceId(123)
                .type(StepType.NOOP)
                .build(),
            "maestro_workflow_runtime_summary",
            Collections.singletonMap("instance_status", "CREATED")));

    Assert.assertTrue(endTask.execute(workflow, testTask, null));

    verify(rollupAggregationHelper, times(1)).calculateRollupBase(any());
    // verify rollupBase set
    Assert.assertNotNull(
        ((WorkflowRuntimeSummary) testTask.getOutputData().get("maestro_workflow_runtime_summary"))
            .getRollupBase());
    Assert.assertEquals(
        1,
        ((WorkflowRuntimeSummary) testTask.getOutputData().get("maestro_workflow_runtime_summary"))
            .getRollupBase()
            .getTotalLeafCount());
    Assert.assertEquals(
        1,
        ((WorkflowRuntimeSummary) testTask.getOutputData().get("maestro_workflow_runtime_summary"))
            .getRollupBase()
            .getOverview()
            .size());

    verify(publisher, times(1)).publish(any(WorkflowInstanceUpdateJobEvent.class));
    WorkflowRuntimeSummary runtimeSummary =
        (WorkflowRuntimeSummary) testTask.getOutputData().get("maestro_workflow_runtime_summary");
    Assert.assertEquals(WorkflowInstance.Status.IN_PROGRESS, runtimeSummary.getInstanceStatus());
    Assert.assertEquals(123456L, runtimeSummary.getStartTime().longValue());
    Assert.assertEquals(1, runtimeSummary.getTimeline().getTimelineEvents().size());
    Assert.assertEquals(
        TimelineLogEvent.builder()
            .timestamp(234567L)
            .message("Workflow instance is dequeued.")
            .level(TimelineEvent.Level.INFO)
            .build(),
        runtimeSummary.getTimeline().getTimelineEvents().get(0));
  }