WorkItemStatusClient { public synchronized WorkItemServiceState reportError(Throwable e) throws IOException { checkState(!finalStateSent, "cannot reportUpdates after sending a final state"); WorkItemStatus status = createStatusUpdate(true); Throwable t = e instanceof UserCodeException ? e.getCause() : e; Status error = new Status(); error.setCode(2); String logPrefix = String.format("Failure processing work item %s", uniqueWorkId()); if (isOutOfMemoryError(t)) { String message = "An OutOfMemoryException occurred. Consider specifying higher memory " + "instances in PipelineOptions.\n"; LOG.error("{}: {}", logPrefix, message); error.setMessage(message + DataflowWorkerLoggingHandler.formatException(t)); } else { LOG.error( "{}: Uncaught exception occurred during work unit execution. This will be retried.", logPrefix, t); error.setMessage(DataflowWorkerLoggingHandler.formatException(t)); } status.setErrors(ImmutableList.of(error)); 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 reportError() throws IOException { RuntimeException error = new RuntimeException(); error.fillInStackTrace(); statusClient.reportError(error); verify(workUnitClient).reportWorkItemStatus(statusCaptor.capture()); WorkItemStatus workStatus = statusCaptor.getValue(); assertThat(workStatus.getWorkItemId(), equalTo(Long.toString(WORK_ID))); assertThat(workStatus.getCompleted(), equalTo(true)); assertThat(workStatus.getReportIndex(), equalTo(INITIAL_REPORT_INDEX)); assertThat(workStatus.getErrors(), hasSize(1)); Status status = workStatus.getErrors().get(0); assertThat(status.getCode(), equalTo(2)); assertThat(status.getMessage(), containsString("WorkItemStatusClientTest")); }