BatchDataflowWorker implements Closeable { public boolean getAndPerformWork() throws IOException { while (true) { Optional<WorkItem> work = workUnitClient.getWorkItem(); if (work.isPresent()) { WorkItemStatusClient statusProvider = new WorkItemStatusClient(workUnitClient, work.get()); return doWork(work.get(), statusProvider); } } } protected  BatchDataflowWorker(
      RunnerApi.@Nullable Pipeline pipeline,
      SdkHarnessRegistry sdkHarnessRegistry,
      WorkUnitClient workUnitClient,
      DataflowMapTaskExecutorFactory mapTaskExecutorFactory,
      DataflowWorkerHarnessOptions options); boolean getAndPerformWork(); void startStatusServer(); @Override void close();  }
@Test public void testWhenNoWorkIsReturnedThatWeImmediatelyRetry() throws Exception { final String workItemId = "14"; BatchDataflowWorker worker = new BatchDataflowWorker( null , SdkHarnessRegistries.emptySdkHarnessRegistry(), mockWorkUnitClient, IntrinsicMapTaskExecutorFactory.defaultFactory(), options); WorkItem workItem = new WorkItem(); workItem.setId(Long.parseLong(workItemId)); workItem.setJobId("SuccessfulEmptyMapTask"); workItem.setInitialReportIndex(12L); workItem.setMapTask( new MapTask() .setInstructions(new ArrayList<ParallelInstruction>()) .setStageName("testStage")); workItem.setLeaseExpireTime(TimeUtil.toCloudTime(Instant.now())); workItem.setReportStatusInterval(TimeUtil.toCloudDuration(Duration.standardMinutes(1))); when(mockWorkUnitClient.getWorkItem()) .thenReturn(Optional.<WorkItem>absent()) .thenReturn(Optional.of(workItem)); assertTrue(worker.getAndPerformWork()); verify(mockWorkUnitClient) .reportWorkItemStatus( MockitoHamcrest.argThat( new TypeSafeMatcher<WorkItemStatus>() { @Override public void describeTo(Description description) {} @Override protected boolean matchesSafely(WorkItemStatus item) { assertTrue(item.getCompleted()); assertEquals(workItemId, item.getWorkItemId()); return true; } })); }