protected Optional observeDriverTermination()

in spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/observers/BaseAppDriverObserver.java [79:143]
60 lines of code
13 McCabe index (conditional complexity)

  protected Optional<ApplicationState> observeDriverTermination(
      final Pod driverPod, final boolean driverReady, final ApplicationSpec spec) {
    PodStatus status = driverPod.getStatus();
    if (status == null
        || status.getContainerStatuses() == null
        || status.getContainerStatuses().isEmpty()) {
      log.warn("Cannot determine driver pod status, the pod may in pending state.");
      return Optional.empty();
    }

    if (PodPhase.FAILED.equals(PodPhase.getPhase(driverPod))) {
      ApplicationState state = new ApplicationState(Failed, DRIVER_FAILED_MESSAGE);
      if ("Evicted".equalsIgnoreCase(status.getReason())) {
        state = new ApplicationState(ApplicationStateSummary.DriverEvicted, DRIVER_FAILED_MESSAGE);
      }
      state.setLastObservedDriverStatus(status);
      return Optional.of(state);
    }

    if (PodPhase.SUCCEEDED.equals(PodPhase.getPhase(driverPod))) {
      ApplicationState state;
      if (driverReady) {
        state = new ApplicationState(Succeeded, DRIVER_COMPLETED_MESSAGE);
      } else {
        state = new ApplicationState(Failed, DRIVER_TERMINATED_BEFORE_INITIALIZATION_MESSAGE);
        state.setLastObservedDriverStatus(status);
      }
      return Optional.of(state);
    }

    List<ContainerStatus> initContainerStatusList = status.getInitContainerStatuses();
    if (initContainerStatusList != null
        && initContainerStatusList.parallelStream().anyMatch(PodUtils::isContainerFailed)) {
      ApplicationState applicationState =
          new ApplicationState(Failed, DRIVER_FAILED_INIT_CONTAINERS_MESSAGE);
      applicationState.setLastObservedDriverStatus(status);
      return Optional.of(applicationState);
    }
    List<ContainerStatus> containerStatusList = status.getContainerStatuses();
    List<ContainerStatus> terminatedCriticalContainers =
        ModelUtils.findDriverMainContainerStatus(spec, containerStatusList).stream()
            .filter(PodUtils::isContainerTerminated)
            .toList();

    if (!terminatedCriticalContainers.isEmpty()) {
      ApplicationState state;
      if (terminatedCriticalContainers.parallelStream().anyMatch(PodUtils::isContainerFailed)) {
        state = new ApplicationState(Failed, DRIVER_FAILED_MESSAGE);
      } else {
        state = new ApplicationState(Succeeded, DRIVER_SUCCEEDED_MESSAGE);
      }
      state.setLastObservedDriverStatus(status);
      return Optional.of(state);
    }

    boolean isDriverRestarted =
        ModelUtils.findDriverMainContainerStatus(spec, containerStatusList).stream()
            .anyMatch(PodUtils::isContainerRestarted);
    if (isDriverRestarted) {
      ApplicationState state = new ApplicationState(Failed, DRIVER_RESTARTED_MESSAGE);
      state.setLastObservedDriverStatus(status);
      return Optional.of(state);
    }
    return Optional.empty();
  }