public synchronized @Nullable WorkItemServiceState reportError(Throwable e) throws IOException {
    checkState(!finalStateSent, "cannot reportUpdates after sending a final state");
    if (wasAskedToAbort) {
      LOG.info("Service already asked to abort work item, not reporting ignored progress.");
      return null;
    }
    WorkItemStatus status = createStatusUpdate(true);

    // TODO: Provide more structure representation of error, e.g., the serialized exception object.
    // TODO: Look into moving the stack trace thinning into the client.
    Throwable t = e instanceof UserCodeException ? e.getCause() : e;

    Status error = new Status();
    error.setCode(2); // Code.UNKNOWN.  TODO: Replace with a generated definition.
    // TODO: Attach the stack trace as exception details, not to the message.
    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 if (isReadLoopAbortedError(t)) {
      LOG.debug("Read loop aborted error occurred during work unit execution", 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);
  }
@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"));
  }