code: public static void finalizeFlow(final ExecutorLoader executorLoader, final AlerterHolder
        alerterHolder, final ExecutableFlow flow, final String reason,
        @Nullable final Throwable originalError) {
      boolean alertUser = true;
  
      // First check if the execution in the datastore is finished.
      try {
        final ExecutableFlow dsFlow;
        if (isFinished(flow)) {
          dsFlow = flow;
        } else {
          dsFlow = executorLoader.fetchExecutableFlow(flow.getExecutionId());
  
          // If it's marked finished, we're good. If not, we fail everything and then mark it
          // finished.
          if (!isFinished(dsFlow)) {
            failEverything(dsFlow);
            executorLoader.updateExecutableFlow(dsFlow);
          }
        }
  
        if (flow.getEndTime() == -1) {
          flow.setEndTime(System.currentTimeMillis());
          executorLoader.updateExecutableFlow(dsFlow);
        }
      } catch (final ExecutorManagerException e) {
        // If failed due to azkaban internal error, do not alert user.
        alertUser = false;
        logger.error("Failed to finalize flow " + flow.getExecutionId() + ", do not alert user.", e);
      }
  
      if (alertUser) {
        alertUserOnFlowFinished(flow, alerterHolder, getFinalizeFlowReasons(reason, originalError));
      }
    }
docs: If the current status of the execution is not one of the finished statuses, mark the execution
  as failed in the DB.
  
  @param executorLoader the executor loader
  @param alerterHolder the alerter holder
  @param flow the execution
  @param reason reason for finalizing the execution
  @param originalError the cause, if execution is being finalized because of an error