State getStateWithRetries(BackOff attempts, Sleeper sleeper) throws IOException {
    if (terminalState != null) {
      return terminalState;
    }
    Job job = getJobWithRetries(attempts, sleeper);
    latestStateString = job.getCurrentState();
    return MonitoringUtil.toState(latestStateString);
  }
@Test
  public void testGetStateWithRetriesPassesExceptionThrough() 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());

    thrown.expect(IOException.class);
    job.getStateWithRetries(
        BackOffAdapter.toGcpBackOff(DataflowPipelineJob.STATUS_BACKOFF_FACTORY.backoff()),
        fastClock::sleep);
  }