protected DAGStatus getDAGStatusViaRM()

in tez-api/src/main/java/org/apache/tez/dag/api/client/DAGClientImpl.java [477:559]
79 lines of code
24 McCabe index (conditional complexity)

  protected DAGStatus getDAGStatusViaRM() throws TezException, IOException {
    LOG.debug("Get DAG status via framework client for app: {} dag: {}", appId, dagId);
    ApplicationReport appReport;
    try {
      appReport = frameworkClient.getApplicationReport(appId);
      LOG.debug("Got appReport from framework client: {}", appReport);
    } catch (ApplicationNotFoundException e) {
      LOG.info("DAG is no longer running - application not found by YARN", e);
      throw new DAGNotRunningException(e);
    } catch (YarnException e) {
      throw new TezException(e);
    }

    if(appReport == null) {
      throw new TezException("Unknown/Invalid appId: " + appId);
    }

    DAGProtos.DAGStatusProto.Builder builder = DAGProtos.DAGStatusProto.newBuilder();
    DAGStatus dagStatus = new DAGStatus(builder, DagStatusSource.RM);
    DAGProtos.DAGStatusStateProto dagState;
    switch (appReport.getYarnApplicationState()) {
      case NEW:
      case NEW_SAVING:
      case SUBMITTED:
      case ACCEPTED:
        dagState = DAGProtos.DAGStatusStateProto.DAG_SUBMITTED;
        break;
      case RUNNING:
        dagState = DAGProtos.DAGStatusStateProto.DAG_RUNNING;
        break;
      case FAILED:
        dagState = DAGProtos.DAGStatusStateProto.DAG_FAILED;
        break;
      case KILLED:
        dagState = DAGProtos.DAGStatusStateProto.DAG_KILLED;
        break;
      case FINISHED:
        switch(appReport.getFinalApplicationStatus()) {
          case UNDEFINED:
          case FAILED:
            dagState = DAGProtos.DAGStatusStateProto.DAG_FAILED;
            break;
          case KILLED:
            dagState = DAGProtos.DAGStatusStateProto.DAG_KILLED;
            break;
          case SUCCEEDED:
            dagState = DAGProtos.DAGStatusStateProto.DAG_SUCCEEDED;
            break;
          default:
            throw new TezUncheckedException("Encountered unknown final application"
                + " status from YARN"
                + ", appState=" + appReport.getYarnApplicationState()
                + ", finalStatus=" + appReport.getFinalApplicationStatus());
        }
        break;
      default:
        throw new TezUncheckedException("Encountered unknown application state"
            + " from YARN, appState=" + appReport.getYarnApplicationState());
    }

    builder.setState(dagState);
    // workaround before YARN-2560 is fixed
    if (appReport.getFinalApplicationStatus() == FinalApplicationStatus.FAILED
        || appReport.getFinalApplicationStatus() == FinalApplicationStatus.KILLED) {
      long startTime = System.currentTimeMillis();
      while((appReport.getDiagnostics() == null
          || appReport.getDiagnostics().isEmpty())
          && (System.currentTimeMillis() - startTime) < diagnoticsWaitTimeout) {
        try {
          Thread.sleep(100);
          appReport = frameworkClient.getApplicationReport(appId);
        } catch (YarnException e) {
          throw new TezException(e);
        } catch (InterruptedException e) {
          throw new TezException(e);
        }
      }
    }
    if (appReport.getDiagnostics() != null) {
      builder.addAllDiagnostics(Collections.singleton(appReport.getDiagnostics()));
    }
    return dagStatus;
  }