@Override
  public @Nullable State waitUntilFinish() {
    return waitUntilFinish(Duration.millis(-1));
  }
@Test
  public void testWaitToFinishTimeFail() 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();
    State state =
        job.waitUntilFinish(Duration.millis(4), null, fastClock::sleep, fastClock::nanoTime);
    assertEquals(null, state);
    long timeDiff = TimeUnit.NANOSECONDS.toMillis(fastClock.nanoTime() - startTime);
    // Should only have slept for the 4 ms allowed.
    assertEquals(4L, timeDiff);
  }