Cleanup zombie jobs that occur after restart (airbytehq#1792)

airbyte-scheduler/src/main/java/io/airbyte/scheduler/SchedulerApp.java

@@ -85,7 +85,7 @@ public SchedulerApp(Path workspaceRoot,
     this.configRepository = configRepository;
   }
 
-  public void start() {
+  public void start() throws IOException {
     final ExecutorService workerThreadPool = Executors.newFixedThreadPool(MAX_WORKERS, THREAD_FACTORY);
     final ScheduledExecutorService scheduledPool = Executors.newSingleThreadScheduledExecutor();
     final WorkerRunFactory workerRunFactory = new WorkerRunFactory(workspaceRoot, pbf);
@@ -94,6 +94,10 @@ public void start() {
     final JobScheduler jobScheduler = new JobScheduler(jobPersistence, configRepository);
     final JobSubmitter jobSubmitter = new JobSubmitter(workerThreadPool, jobPersistence, configRepository, workerRunFactory);
 
+    // We cancel jobs that where running before the restart. They are not being monitored by the worker
+    // anymore.
+    cleanupZombies(jobPersistence);
+
     scheduledPool.scheduleWithFixedDelay(
         () -> {
           jobRetrier.run();
@@ -107,6 +111,12 @@ public void start() {
     Runtime.getRuntime().addShutdownHook(new GracefulShutdownHandler(Duration.ofSeconds(GRACEFUL_SHUTDOWN_SECONDS), workerThreadPool, scheduledPool));
   }
 
+  private void cleanupZombies(JobPersistence jobPersistence) throws IOException {
+    for (Job zombieJob : jobPersistence.listJobsWithStatus(JobStatus.RUNNING)) {
+      jobPersistence.cancelJob(zombieJob.getId());
+    }
+  }
+
   private static ProcessBuilderFactory getProcessBuilderFactory(Configs configs) {
     if (configs.getWorkerEnvironment() == Configs.WorkerEnvironment.KUBERNETES) {
       return new KubeProcessBuilderFactory(configs.getWorkspaceRoot());