DataflowPipelineJob implements PipelineResult { @VisibleForTesting State getStateWithRetriesOrUnknownOnException(BackOff attempts, Sleeper sleeper) { try { return getStateWithRetries(attempts, sleeper); } catch (IOException exn) { return State.UNKNOWN; } } 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 testGetStateReturnsServiceState() throws Exception { Dataflow.Projects.Locations.Jobs.Get statusRequest = mock(Dataflow.Projects.Locations.Jobs.Get.class); Job statusResponse = new Job(); statusResponse.setCurrentState("JOB_STATE_" + State.RUNNING.name()); when(mockJobs.get(eq(PROJECT_ID), eq(REGION_ID), eq(JOB_ID))).thenReturn(statusRequest); when(statusRequest.execute()).thenReturn(statusResponse); DataflowPipelineJob job = new DataflowPipelineJob(DataflowClient.create(options), JOB_ID, options, ImmutableMap.of()); assertEquals( State.RUNNING, job.getStateWithRetriesOrUnknownOnException( BackOffAdapter.toGcpBackOff(DataflowPipelineJob.STATUS_BACKOFF_FACTORY.backoff()), fastClock)); }
@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)); long timeDiff = TimeUnit.NANOSECONDS.toMillis(fastClock.nanoTime() - startTime); checkValidInterval( DataflowPipelineJob.STATUS_POLLING_INTERVAL, DataflowPipelineJob.STATUS_POLLING_RETRIES, timeDiff); }