WorkItemStatusClient { public synchronized WorkItemServiceState reportUpdate( @Nullable DynamicSplitResult dynamicSplitResult, Duration requestedLeaseDuration) throws Exception { checkState(worker != null, "setWorker should be called before reportUpdate"); checkState(!finalStateSent, "cannot reportUpdates after sending a final state"); checkArgument(requestedLeaseDuration != null, "requestLeaseDuration must be non-null"); WorkItemStatus status = createStatusUpdate(false); status.setRequestedLeaseDuration(TimeUtil.toCloudDuration(requestedLeaseDuration)); populateProgress(status); populateSplitResult(status, dynamicSplitResult); return execute(status); } WorkItemStatusClient(WorkUnitClient workUnitClient, WorkItem workItem); String uniqueWorkId(); synchronized void setWorker(
      DataflowWorkExecutor worker, BatchModeExecutionContext executionContext); synchronized WorkItemServiceState reportError(Throwable e); synchronized WorkItemServiceState reportSuccess(); synchronized WorkItemServiceState reportUpdate(
      @Nullable DynamicSplitResult dynamicSplitResult, Duration requestedLeaseDuration); Iterable<CounterUpdate> extractMsecCounters(boolean isFinalUpdate); long extractThrottleTime();  }
@Test public void reportUpdate() throws Exception { when(worker.extractMetricUpdates()).thenReturn(Collections.emptyList()); statusClient.setWorker(worker, executionContext); statusClient.reportUpdate(null, LEASE_DURATION); verify(workUnitClient).reportWorkItemStatus(statusCaptor.capture()); WorkItemStatus workStatus = statusCaptor.getValue(); assertThat(workStatus.getCompleted(), equalTo(false)); }
@Test public void reportUpdateBeforeSetWorker() throws Exception { thrown.expect(IllegalStateException.class); thrown.expectMessage("setWorker"); thrown.expectMessage("reportUpdate"); statusClient.reportUpdate(null, null); }