public synchronized @Nullable WorkItemServiceState reportSuccess() throws IOException {
    checkState(!finalStateSent, "cannot reportSuccess after sending a final state");
    checkState(worker != null, "setWorker should be called before reportSuccess");
    if (wasAskedToAbort) {
      LOG.info("Service already asked to abort work item, not reporting ignored progress.");
      return null;
    }

    WorkItemStatus status = createStatusUpdate(true);

    if (worker instanceof SourceOperationExecutor) {
      // TODO: Find out a generic way for the DataflowWorkExecutor to report work-specific results
      // into the work update.
      SourceOperationResponse response = ((SourceOperationExecutor) worker).getResponse();
      if (response != null) {
        status.setSourceOperationResponse(response);
      }
    }

    LOG.info("Success processing work item {}", uniqueWorkId());

    return execute(status);
  }
@Test
  public void reportSuccess() throws IOException {
    when(worker.extractMetricUpdates()).thenReturn(Collections.emptyList());
    statusClient.setWorker(worker, executionContext);

    statusClient.reportSuccess();
    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(), nullValue());
  }