private void getFromYarnRestApi()

in streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/task/FlinkHttpWatcher.java [487:555]
62 lines of code
17 McCabe index (conditional complexity)

  private void getFromYarnRestApi(Application application, StopFrom stopFrom) throws Exception {
    log.debug("FlinkHttpWatcher getFromYarnRestApi starting...");
    OptionState optionState = OPTIONING.get(application.getId());

    /*
     If the status of the last time is CANCELING (flink rest server is not closed at the time of getting information)
     and the status is not obtained this time (flink rest server is closed),
     the task is considered CANCELED
    */
    Byte flag = CANCELING_CACHE.getIfPresent(application.getId());
    if (flag != null) {
      log.info("FlinkHttpWatcher previous state: canceling.");
      if (StopFrom.NONE.equals(stopFrom)) {
        log.error(
            "FlinkHttpWatcher query previous state was canceling and stopFrom NotFound,savePoint expired!");
        savePointService.expire(application.getId());
      }
      application.setState(FlinkAppState.CANCELED.getValue());
      cleanSavepoint(application);
      cleanOptioning(optionState, application.getId());
      doPersistMetrics(application, true);
    } else {
      // query the status from the yarn rest Api
      YarnAppInfo yarnAppInfo = httpYarnAppInfo(application);
      if (yarnAppInfo == null) {
        if (!ExecutionMode.REMOTE.equals(application.getExecutionModeEnum())) {
          throw new RuntimeException("FlinkHttpWatcher getFromYarnRestApi failed ");
        }
      } else {
        try {
          String state = yarnAppInfo.getApp().getFinalStatus();
          FlinkAppState flinkAppState = FlinkAppState.of(state);
          if (FlinkAppState.OTHER.equals(flinkAppState)) {
            return;
          }
          if (FlinkAppState.KILLED.equals(flinkAppState)) {
            if (StopFrom.NONE.equals(stopFrom)) {
              log.error(
                  "FlinkHttpWatcher getFromYarnRestApi,job was killed and stopFrom NotFound,savePoint expired!");
              savePointService.expire(application.getId());
            }
            flinkAppState = FlinkAppState.CANCELED;
            cleanSavepoint(application);
            application.setEndTime(new Date());
          }
          if (FlinkAppState.SUCCEEDED.equals(flinkAppState)) {
            flinkAppState = FlinkAppState.FINISHED;
          }
          application.setState(flinkAppState.getValue());
          cleanOptioning(optionState, application.getId());
          doPersistMetrics(application, true);
          if (flinkAppState.equals(FlinkAppState.FAILED)
              || flinkAppState.equals(FlinkAppState.LOST)
              || (flinkAppState.equals(FlinkAppState.CANCELED) && StopFrom.NONE.equals(stopFrom))
              || applicationService.checkAlter(application)) {
            doAlert(application, flinkAppState);
            stopCanceledJob(application.getId());
            if (flinkAppState.equals(FlinkAppState.FAILED)) {
              applicationService.start(application, true);
            }
          }
        } catch (Exception e) {
          if (!ExecutionMode.REMOTE.equals(application.getExecutionModeEnum())) {
            throw new RuntimeException("FlinkHttpWatcher getFromYarnRestApi error,", e);
          }
        }
      }
    }
  }