@Override
  public void onWorkflowFinalized(Workflow workflow) {
    WorkflowSummary summary = StepHelper.retrieveWorkflowSummary(objectMapper, workflow.getInput());
    WorkflowRuntimeSummary runtimeSummary = retrieveWorkflowRuntimeSummary(workflow);
    String reason = workflow.getReasonForIncompletion();
    LOG.info(
        "Workflow {} with execution_id [{}] is finalized with internal state [{}] and reason [{}]",
        summary.getIdentity(),
        workflow.getWorkflowId(),
        workflow.getStatus(),
        reason);
    metrics.counter(
        MetricConstants.WORKFLOW_STATUS_LISTENER_CALL_BACK_METRIC,
        getClass(),
        TYPE_TAG,
        "onWorkflowFinalized",
        MetricConstants.STATUS_TAG,
        workflow.getStatus().name());

    if (reason != null
        && workflow.getStatus() == Workflow.WorkflowStatus.FAILED
        && reason.startsWith(MaestroStartTask.DEDUP_FAILURE_PREFIX)) {
      LOG.info(
          "Workflow {} with execution_id [{}] has not actually started, thus skip onWorkflowFinalized.",
          summary.getIdentity(),
          workflow.getWorkflowId());
      return; // special case doing nothing
    }

    WorkflowInstance.Status instanceStatus =
        instanceDao.getWorkflowInstanceStatus(
            summary.getWorkflowId(), summary.getWorkflowInstanceId(), summary.getWorkflowRunId());
    if (instanceStatus == null
        || (instanceStatus.isTerminal() && workflow.getStatus().isTerminal())) {
      LOG.info(
          "Workflow {} with execution_id [{}] does not exist or already "
              + "in a terminal state [{}] with internal state [{}], thus skip onWorkflowFinalized.",
          summary.getIdentity(),
          workflow.getWorkflowId(),
          instanceStatus,
          workflow.getStatus());
      return;
    }

    Map<String, Task> realTaskMap = TaskHelper.getUserDefinedRealTaskMap(workflow);
    // cancel internally failed tasks
    realTaskMap.values().stream()
        .filter(task -> !StepHelper.retrieveStepStatus(task.getOutputData()).isTerminal())
        .forEach(task -> maestroTask.cancel(workflow, task, null));

    WorkflowRuntimeOverview overview =
        TaskHelper.computeOverview(
            objectMapper, summary, runtimeSummary.getRollupBase(), realTaskMap);

    try {
      validateAndUpdateOverview(overview, summary);
      switch (workflow.getStatus()) {
        case TERMINATED: // stopped due to stop request
          if (reason != null && reason.startsWith(FAILURE_REASON_PREFIX)) {
            update(workflow, WorkflowInstance.Status.FAILED, summary, overview);
          } else {
            update(workflow, WorkflowInstance.Status.STOPPED, summary, overview);
          }
          break;
        case TIMED_OUT:
          update(workflow, WorkflowInstance.Status.TIMED_OUT, summary, overview);
          break;
        default: // other status (FAILED, COMPLETED, PAUSED, RUNNING) to be handled here.
          Optional<Task.Status> done =
              TaskHelper.checkProgress(realTaskMap, summary, overview, true);
          switch (done.orElse(Task.Status.IN_PROGRESS)) {
              /**
               * This is a special status to indicate that the workflow has succeeded. Check {@link
               * TaskHelper#checkProgress} for more details.
               */
            case FAILED_WITH_TERMINAL_ERROR:
              WorkflowInstance.Status nextStatus =
                  AggregatedViewHelper.deriveAggregatedStatus(
                      instanceDao, summary, WorkflowInstance.Status.SUCCEEDED, overview);
              if (!nextStatus.isTerminal()) {
                throw new MaestroInternalError(
                    "Invalid status: [%s], expecting a terminal one", nextStatus);
              }
              update(workflow, nextStatus, summary, overview);
              break;
            case FAILED:
            case CANCELED: // due to step failure
              update(workflow, WorkflowInstance.Status.FAILED, summary, overview);
              break;
            case TIMED_OUT:
              update(workflow, WorkflowInstance.Status.TIMED_OUT, summary, overview);
              break;
              // all other status are invalid
            default:
              metrics.counter(
                  MetricConstants.WORKFLOW_STATUS_LISTENER_CALL_BACK_METRIC,
                  getClass(),
                  TYPE_TAG,
                  "invalidStatusOnWorkflowFinalized");
              throw new MaestroInternalError(
                  "Invalid status [%s] onWorkflowFinalized", workflow.getStatus());
          }
          break;
      }
    } catch (MaestroInternalError | IllegalArgumentException e) {
      // non-retryable error and still fail the instance
      LOG.warn("onWorkflowFinalized is failed with a non-retryable error", e);
      metrics.counter(
          MetricConstants.WORKFLOW_STATUS_LISTENER_CALL_BACK_METRIC,
          getClass(),
          TYPE_TAG,
          "nonRetryableErrorOnWorkflowFinalized");
      update(
          workflow,
          WorkflowInstance.Status.FAILED,
          summary,
          overview,
          Details.create(
              e.getMessage(), "onWorkflowFinalized is failed with non-retryable error."));
    }
  }
@Test
  public void testInconsistentStatsOnWorkflowFinalized() {
    StepRuntimeState state = new StepRuntimeState();
    state.setStatus(StepInstance.Status.FATALLY_FAILED);
    when(stepInstanceDao.getAllStepStates(any(), anyLong(), anyLong()))
        .thenReturn(singletonMap("foo", state));
    when(workflow.getStatus()).thenReturn(Workflow.WorkflowStatus.TERMINATED);
    when(instanceDao.getWorkflowInstanceStatus(eq("test-workflow-id"), anyLong(), anyLong()))
        .thenReturn(WorkflowInstance.Status.IN_PROGRESS);
    statusListener.onWorkflowFinalized(workflow);
    Assert.assertEquals(
        1L,
        metricRepo
            .getCounter(
                MetricConstants.WORKFLOW_STATUS_LISTENER_CALL_BACK_METRIC,
                MaestroWorkflowStatusListener.class,
                "type",
                "inconsistentStatsOnWorkflowFinalized")
            .count());
    ArgumentCaptor<Timeline> timelineCaptor = ArgumentCaptor.forClass(Timeline.class);
    verify(instanceDao, times(1))
        .updateWorkflowInstance(
            any(), any(), timelineCaptor.capture(), eq(WorkflowInstance.Status.STOPPED), anyLong());
    Timeline timeline = timelineCaptor.getValue();
    Assert.assertEquals(1, timeline.getTimelineEvents().size());
    Assert.assertEquals(
        "Workflow instance status is updated to [STOPPED] due to [test-reason]",
        timeline.getTimelineEvents().get(0).getMessage());
    verify(publisher, times(1)).publishOrThrow(any(), any());
  }