public ApplicationStatus terminateOrRestart()

in spark-operator-api/src/main/java/org/apache/spark/k8s/operator/status/ApplicationStatus.java [78:154]
66 lines of code
13 McCabe index (conditional complexity)

  public ApplicationStatus terminateOrRestart(
      final RestartConfig restartConfig,
      final ResourceRetainPolicy resourceRetainPolicy,
      String stateMessageOverride,
      boolean trimStateTransitionHistory) {
    if (!currentState.currentStateSummary.isStopping()) {
      // application is not stopping, skip
      throw new IllegalStateException(
          "Spark application cannot be directly terminated unless in stopping "
              + "state, current state is: "
              + currentState);
    }

    if (!RestartPolicy.attemptRestartOnState(
        restartConfig.getRestartPolicy(), currentState.getCurrentStateSummary())) {
      // no restart configured
      ApplicationState state =
          new ApplicationState(ApplicationStateSummary.ResourceReleased, stateMessageOverride);
      if (ResourceRetainPolicy.Always.equals(resourceRetainPolicy)
          || ResourceRetainPolicy.OnFailure.equals(resourceRetainPolicy)
              && currentState.currentStateSummary.isFailure()) {
        state = terminateAppWithoutReleaseResource(stateMessageOverride);
      }
      return new ApplicationStatus(
          state,
          createUpdatedHistoryWithNewState(state),
          previousAttemptSummary,
          currentAttemptSummary);
    }

    if (currentAttemptSummary.getAttemptInfo().getId() >= restartConfig.getMaxRestartAttempts()) {
      String stateMessage =
          String.format(EXCEED_MAX_RETRY_ATTEMPT_MESSAGE, restartConfig.getMaxRestartAttempts());
      if (stateMessageOverride != null && !stateMessageOverride.isEmpty()) {
        stateMessage += stateMessageOverride;
      }
      // max number of restart attempt reached
      ApplicationState state =
          new ApplicationState(ApplicationStateSummary.ResourceReleased, stateMessage);
      if (ResourceRetainPolicy.Always.equals(resourceRetainPolicy)
          || ResourceRetainPolicy.OnFailure.equals(resourceRetainPolicy)
              && currentState.currentStateSummary.isFailure()) {
        state = terminateAppWithoutReleaseResource(stateMessage);
      }
      // still use previous & current attempt summary - they are to be updated only upon
      // new restart
      return new ApplicationStatus(
          state,
          createUpdatedHistoryWithNewState(state),
          previousAttemptSummary,
          currentAttemptSummary);
    }

    AttemptInfo nextAttemptInfo = currentAttemptSummary.getAttemptInfo().createNextAttemptInfo();
    ApplicationAttemptSummary nextAttemptSummary = new ApplicationAttemptSummary(nextAttemptInfo);
    ApplicationState state =
        new ApplicationState(ApplicationStateSummary.ScheduledToRestart, stateMessageOverride);

    if (trimStateTransitionHistory) {
      // when truncating, put all previous history entries into previous attempt summary
      ApplicationAttemptSummary newPrevSummary =
          new ApplicationAttemptSummary(
              currentAttemptSummary.getAttemptInfo(), stateTransitionHistory);
      return new ApplicationStatus(
          state,
          Collections.singletonMap(getCurrentStateId() + 1, state),
          newPrevSummary,
          nextAttemptSummary);
    } else {
      // when truncating is disabled, currentAttempt becomes the new 'previous'
      return new ApplicationStatus(
          state,
          createUpdatedHistoryWithNewState(state),
          currentAttemptSummary,
          nextAttemptSummary);
    }
  }