public static State toState(@Nullable String stateName) {
    if (stateName == null) {
      return State.UNRECOGNIZED;
    }

    switch (stateName) {
      case "JOB_STATE_UNKNOWN":
        return State.UNKNOWN;
      case "JOB_STATE_STOPPED":
        return State.STOPPED;
      case "JOB_STATE_FAILED":
        return State.FAILED;
      case "JOB_STATE_CANCELLED":
        return State.CANCELLED;
      case "JOB_STATE_UPDATED":
        return State.UPDATED;

      case "JOB_STATE_RUNNING":
      case "JOB_STATE_PENDING": // Job has not yet started; closest mapping is RUNNING
      case "JOB_STATE_DRAINING": // Job is still active; the closest mapping is RUNNING
      case "JOB_STATE_CANCELLING": // Job is still active; the closest mapping is RUNNING
      case "JOB_STATE_RESOURCE_CLEANING_UP": // Job is still active; the closest mapping is RUNNING
        return State.RUNNING;

      case "JOB_STATE_DONE":
      case "JOB_STATE_DRAINED": // Job has successfully terminated; closest mapping is DONE
        return State.DONE;
      default:
        LOG.warn(
            "Unrecognized state from Dataflow service: {}."
                + " This is likely due to using an older version of Beam.",
            stateName);
        return State.UNRECOGNIZED;
    }
  }
@Test
  public void testToStateNormal() {
    // Trivially mapped cases
    assertEquals(State.UNKNOWN, MonitoringUtil.toState("JOB_STATE_UNKNOWN"));
    assertEquals(State.STOPPED, MonitoringUtil.toState("JOB_STATE_STOPPED"));
    assertEquals(State.RUNNING, MonitoringUtil.toState("JOB_STATE_RUNNING"));
    assertEquals(State.DONE, MonitoringUtil.toState("JOB_STATE_DONE"));
    assertEquals(State.FAILED, MonitoringUtil.toState("JOB_STATE_FAILED"));
    assertEquals(State.CANCELLED, MonitoringUtil.toState("JOB_STATE_CANCELLED"));
    assertEquals(State.UPDATED, MonitoringUtil.toState("JOB_STATE_UPDATED"));

    // Non-trivially mapped cases
    assertEquals(State.RUNNING, MonitoringUtil.toState("JOB_STATE_DRAINING"));
    assertEquals(State.DONE, MonitoringUtil.toState("JOB_STATE_DRAINED"));
  }