DataflowPipelineJob implements PipelineResult { State getStateWithRetries(BackOff attempts, Sleeper sleeper) throws IOException { if (terminalState != null) { return terminalState; } Job job = getJobWithRetries(attempts, sleeper); return MonitoringUtil.toState(job.getCurrentState()); } DataflowPipelineJob(
      DataflowClient dataflowClient,
      String jobId,
      DataflowPipelineOptions dataflowOptions,
      Map<AppliedPTransform<?, ?, ?>, String> transformStepNames); String getJobId(); String getProjectId(); DataflowPipelineOptions getDataflowOptions(); String getRegion(); DataflowPipelineJob getReplacedByJob(); @Override @Nullable State waitUntilFinish(); @Override @Nullable State waitUntilFinish(Duration duration); @Nullable @VisibleForTesting State waitUntilFinish(Duration duration, MonitoringUtil.JobMessagesHandler messageHandler); @Override State cancel(); @Override State getState(); @Override MetricResults metrics();  }
@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()); long startTime = fastClock.nanoTime(); thrown.expect(IOException.class); job.getStateWithRetries( BackOffAdapter.toGcpBackOff(DataflowPipelineJob.STATUS_BACKOFF_FACTORY.backoff()), fastClock); }