DirectRunnerJobManager implements JobManager { @Override public Job abortJob(Job job) { DirectJob directJob = jobs.get(job.getExtId()); if (directJob != null) { try { directJob.abort(); } catch (IOException e) { throw new RuntimeException( Strings.lenientFormat("Unable to abort DirectRunner job %s", job.getExtId(), e)); } jobs.remove(job.getExtId()); } return job; } DirectRunnerJobManager(
      DirectRunnerConfigOptions directRunnerConfigOptions,
      DirectJobRegistry jobs,
      MetricsProperties metricsProperties,
      IngestionJobProto.SpecsStreamingUpdateConfig specsStreamingUpdateConfig); @Override Runner getRunnerType(); @Override Job startJob(Job job); @Override Job updateJob(Job job); @Override Job abortJob(Job job); PipelineResult runPipeline(ImportOptions pipelineOptions); @Override Job restartJob(Job job); @Override JobStatus getJobStatus(Job job); @Override List<Job> listRunningJobs();  }
@Test public void shouldAbortJobThenRemoveFromRegistry() throws IOException { Job job = Job.builder() .setSource(source) .setStores(ImmutableMap.of(store.getName(), store)) .setId("id") .build(); job.setExtId("ext1"); job.setStatus(JobStatus.RUNNING); DirectJob directJob = Mockito.mock(DirectJob.class); when(directJobRegistry.get("ext1")).thenReturn(directJob); job = drJobManager.abortJob(job); verify(directJob, times(1)).abort(); verify(directJobRegistry, times(1)).remove("ext1"); }