@VisibleForTesting
  State getStateWithRetriesOrUnknownOnException(BackOff attempts, Sleeper sleeper) {
    try {
      return getStateWithRetries(attempts, sleeper);
    } catch (IOException exn) {
      // The only IOException that getJobWithRetries is permitted to throw is the final IOException
      // that caused the failure of retry. Other exceptions are wrapped in an unchecked exceptions
      // and will propagate.
      return State.UNKNOWN;
    }
  }
@Test
  public void testGetStateNoThrowWithExceptionReturnsUnknown() throws Exception {
    Dataflow.Projects.Locations.Jobs.Get statusRequest =
        mock(Dataflow.Projects.Locations.Jobs.Get.class);

    when(mockJobs.get(eq(PROJECT_ID), eq(REGION_ID), eq(JOB_ID))).thenReturn(statusRequest);
    when(statusRequest.execute()).thenThrow(IOException.class);

    DataflowPipelineJob job =
        new DataflowPipelineJob(DataflowClient.create(options), JOB_ID, options, ImmutableMap.of());

    long startTime = fastClock.nanoTime();
    assertEquals(
        State.UNKNOWN,
        job.getStateWithRetriesOrUnknownOnException(
            BackOffAdapter.toGcpBackOff(DataflowPipelineJob.STATUS_BACKOFF_FACTORY.backoff()),
            fastClock::sleep));
    long timeDiff = TimeUnit.NANOSECONDS.toMillis(fastClock.nanoTime() - startTime);
    checkValidInterval(
        DataflowPipelineJob.STATUS_POLLING_INTERVAL,
        DataflowPipelineJob.STATUS_POLLING_RETRIES,
        timeDiff);
  }