@Override
  public State cancel() throws IOException {
    // Enforce that a cancel() call on the job is done at most once - as
    // a workaround for Dataflow service's current bugs with multiple
    // cancellation, where it may sometimes return an error when cancelling
    // a job that was already cancelled, but still report the job state as
    // RUNNING.
    // To partially work around these issues, we absorb duplicate cancel()
    // calls. This, of course, doesn't address the case when the job terminates
    // externally almost concurrently to calling cancel(), but at least it
    // makes it possible to safely call cancel() multiple times and from
    // multiple threads in one program.
    FutureTask<State> tentativeCancelTask =
        new FutureTask<>(
            () -> {
              Job content = new Job();
              content.setProjectId(getProjectId());
              String currentJobId = getJobId();
              content.setId(currentJobId);
              content.setRequestedState("JOB_STATE_CANCELLED");
              try {
                Job job = dataflowClient.updateJob(currentJobId, content);
                return MonitoringUtil.toState(job.getCurrentState());
              } catch (IOException e) {
                State state = getState();
                if (state.isTerminal()) {
                  LOG.warn("Cancel failed because job is already terminated. State is {}", state);
                  return state;
                } else if (e.getMessage().contains("has terminated")) {
                  // This handles the case where the getState() call above returns RUNNING but the
                  // cancel was rejected because the job is in fact done. Hopefully, someday we can
                  // delete this code if there is better consistency between the State and whether
                  // Cancel succeeds.
                  //
                  // Example message:
                  //    Workflow modification failed. Causes: (7603adc9e9bff51e): Cannot perform
                  //    operation 'cancel' on Job: 2017-04-01_22_50_59-9269855660514862348. Job has
                  //    terminated in state SUCCESS: Workflow job:
                  //    2017-04-01_22_50_59-9269855660514862348 succeeded.
                  LOG.warn("Cancel failed because job is already terminated.", e);
                  return state;
                } else {
                  String errorMsg =
                      String.format(
                          "Failed to cancel job in state %s, "
                              + "please go to the Developers Console to cancel it manually: %s",
                          state,
                          MonitoringUtil.getJobMonitoringPageURL(
                              getProjectId(), getRegion(), getJobId()));
                  LOG.warn(errorMsg);
                  throw new IOException(errorMsg, e);
                }
              }
            });
    if (cancelState.compareAndSet(null, tentativeCancelTask)) {
      // This thread should perform cancellation, while others will
      // only wait for the result.
      cancelState.get().run();
    }
    try {
      return cancelState.get().get();
    } catch (InterruptedException | ExecutionException e) {
      throw new IOException(e);
    }
  }
@Test
  public void testCancelTerminatedJobWithStaleState() throws IOException {
    Dataflow.Projects.Locations.Jobs.Get statusRequest =
        mock(Dataflow.Projects.Locations.Jobs.Get.class);

    Job statusResponse = new Job();
    statusResponse.setCurrentState("JOB_STATE_RUNNING");
    when(mockJobs.get(PROJECT_ID, REGION_ID, JOB_ID)).thenReturn(statusRequest);
    when(statusRequest.execute()).thenReturn(statusResponse);

    Dataflow.Projects.Locations.Jobs.Update update =
        mock(Dataflow.Projects.Locations.Jobs.Update.class);
    when(mockJobs.update(eq(PROJECT_ID), eq(REGION_ID), eq(JOB_ID), any(Job.class)))
        .thenReturn(update);
    when(update.execute()).thenThrow(new IOException("Job has terminated in state SUCCESS"));

    DataflowPipelineJob job =
        new DataflowPipelineJob(DataflowClient.create(options), JOB_ID, options, null);
    State returned = job.cancel();
    assertThat(returned, equalTo(State.RUNNING));
    expectedLogs.verifyWarn("Cancel failed because job is already terminated.");
  }