fail zombie job attempts and add failure reason (#8709)

* fail zombie job attempts and add failure reason

* remove failure reason

Author: lmossman

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

@@ -167,14 +167,20 @@ public void start() throws IOException {
 
   private void cleanupZombies(final JobPersistence jobPersistence, final JobNotifier jobNotifier) throws IOException {
     for (final Job zombieJob : jobPersistence.listJobsWithStatus(JobStatus.RUNNING)) {
-      jobNotifier.failJob("zombie job was cancelled", zombieJob);
+      jobNotifier.failJob("zombie job was failed", zombieJob);
+
+      final int currentAttemptNumber = zombieJob.getAttemptsCount() - 1;
+
       LOGGER.warn(
-          "zombie clean up - job was cancelled. job id: {}, type: {}, scope: {}",
+          "zombie clean up - job attempt was failed. job id: {}, attempt number: {}, type: {}, scope: {}",
           zombieJob.getId(),
+          currentAttemptNumber,
           zombieJob.getConfigType(),
           zombieJob.getScope());
 
-      jobPersistence.cancelJob(zombieJob.getId());
+      jobPersistence.failAttempt(
+          zombieJob.getId(),
+          currentAttemptNumber);
     }
   }
 

airbyte-scheduler/persistence/src/main/java/io/airbyte/scheduler/persistence/JobPersistence.java

@@ -85,8 +85,8 @@ public interface JobPersistence {
   int createAttempt(long jobId, Path logPath) throws IOException;
 
   /**
-   * Sets an attempt to FAILED. Also attempts the parent job to FAILED. The job's status will not be
-   * changed if it is already in a terminal state.
+   * Sets an attempt to FAILED. Also attempts to set the parent job to INCOMPLETE. The job's status
+   * will not be changed if it is already in a terminal state.
    *
    * @param jobId job id
    * @param attemptNumber attempt id
@@ -95,8 +95,8 @@ public interface JobPersistence {
   void failAttempt(long jobId, int attemptNumber) throws IOException;
 
   /**
-   * Sets an attempt to SUCCEEDED. Also attempts the parent job to SUCCEEDED. The job's status is
-   * changed regardless of what state it is in.
+   * Sets an attempt to SUCCEEDED. Also attempts to set the parent job to SUCCEEDED. The job's status
+   * is changed regardless of what state it is in.
    *
    * @param jobId job id
    * @param attemptNumber attempt id