[#12992]feat(core): complete OCC for jobs and job templates

core/src/main/java/org/apache/gravitino/job/JobManager.java

@@ -57,6 +57,9 @@
 import org.apache.gravitino.exceptions.NoSuchEntityException;
 import org.apache.gravitino.exceptions.NoSuchJobException;
 import org.apache.gravitino.exceptions.NoSuchJobTemplateException;
+import org.apache.gravitino.exceptions.NoSuchMetalakeException;
+import org.apache.gravitino.exceptions.NonEmptyEntityException;
+import org.apache.gravitino.exceptions.OptimisticLockException;
 import org.apache.gravitino.json.JsonUtils;
 import org.apache.gravitino.lock.LockType;
 import org.apache.gravitino.lock.TreeLockUtils;
@@ -226,6 +229,8 @@ public void registerJobTemplate(String metalake, JobTemplateEntity jobTemplateEn
             throw new JobTemplateAlreadyExistsException(
                 "Job template with name %s under metalake %s already exists",
                 jobTemplateEntity.name(), metalake);
+          } catch (NoSuchEntityException e) {
+            throw new NoSuchMetalakeException(e, "Metalake %s does not exist", metalake);
           } catch (IOException ioe) {
             throw new RuntimeException(ioe);
           }
@@ -267,44 +272,50 @@ public boolean deleteJobTemplate(String metalake, String jobTemplateName) throws
       return false;
     }
 
-    boolean hasActiveJobs =
-        jobs.stream()
-            .anyMatch(
-                job ->
-                    job.status() != JobHandle.Status.CANCELLED
-                        && job.status() != JobHandle.Status.SUCCEEDED
-                        && job.status() != JobHandle.Status.FAILED);
+    boolean hasActiveJobs = jobs.stream().anyMatch(job -> !isFinishedStatus(job.status()));
     if (hasActiveJobs) {
       throw new InUseException(
           "Job template %s under metalake %s has active jobs associated with it",
           jobTemplateName, metalake);
     }
 
-    // Delete all the job staging directories associated with the job template.
-    String jobTemplateStagingPath =
-        stagingDir.getAbsolutePath() + File.separator + metalake + File.separator + jobTemplateName;
-    File jobTemplateStagingDir = new File(jobTemplateStagingPath);
-    if (jobTemplateStagingDir.exists()) {
+    // Delete the job template entity as well as all the jobs associated with it.
+    boolean deleted =
+        TreeLockUtils.doWithTreeLock(
+            NameIdentifier.of(NamespaceUtil.ofJobTemplate(metalake).levels()),
+            LockType.WRITE,
+            () -> {
+              try {
+                return entityStore.delete(
+                    NameIdentifierUtil.ofJobTemplate(metalake, jobTemplateName),
+                    Entity.EntityType.JOB_TEMPLATE);
+              } catch (NonEmptyEntityException e) {
+                throw new InUseException(
+                    "Job template %s under metalake %s has active jobs associated with it",
+                    jobTemplateName, metalake);
+              } catch (IOException ioe) {
+                throw new RuntimeException(ioe);
+              }
+            });
+    if (!deleted) {
+      return false;
+    }
+
+    // Only remove directories belonging to the observed jobs. A same-name template can be
+    // recreated after the metadata transaction commits, so its parent directory is not ours to
+    // delete.
+    for (JobEntity job : jobs) {
+      String jobStagingPath =
+          stagingDir.getAbsolutePath()
+              + String.format(JOB_STAGING_DIR, metalake, job.jobTemplateName(), job.id());
       try {
-        FileUtils.deleteDirectory(jobTemplateStagingDir);
+        FileUtils.deleteDirectory(new File(jobStagingPath));
       } catch (IOException e) {
-        LOG.error("Failed to delete job template staging directory: {}", jobTemplateStagingPath, e);
+        LOG.error("Failed to delete job staging directory: {}", jobStagingPath, e);
       }
     }
 
-    // Delete the job template entity as well as all the jobs associated with it.
-    return TreeLockUtils.doWithTreeLock(
-        NameIdentifier.of(NamespaceUtil.ofJobTemplate(metalake).levels()),
-        LockType.WRITE,
-        () -> {
-          try {
-            return entityStore.delete(
-                NameIdentifierUtil.ofJobTemplate(metalake, jobTemplateName),
-                Entity.EntityType.JOB_TEMPLATE);
-          } catch (IOException ioe) {
-            throw new RuntimeException(ioe);
-          }
-        });
+    return true;
   }
 
   @Override
@@ -335,9 +346,7 @@ public JobTemplateEntity alterJobTemplate(
                     updateJobTemplateEntity(jobTemplateIdent, jobTemplateEntity, changes));
           } catch (NoSuchEntityException e) {
             throw new NoSuchJobTemplateException(
-                "Job template with name %s under metalake %s does not exist, this could be due to"
-                    + " the job template not existing or updated concurrently. For the latter case"
-                    + " please retry the operation.",
+                "Job template with name %s under metalake %s does not exist",
                 jobTemplateName, metalake);
           } catch (IOException ioe) {
             throw new RuntimeException(ioe);
@@ -490,6 +499,20 @@ public JobEntity runJob(String metalake, String jobTemplateName, Map<String, Str
 
     try {
       entityStore.put(jobEntity, false /* overwrite */);
+    } catch (NoSuchEntityException e) {
+      LOG.error(
+          "Job {} was submitted as execution {} but could not be registered because its template "
+              + "{} or metalake {} no longer exists",
+          jobEntity.name(),
+          jobExecutionId,
+          jobTemplateName,
+          metalake,
+          e);
+      throw new NoSuchJobTemplateException(
+          e,
+          "Job template with name %s under metalake %s does not exist",
+          jobTemplateName,
+          metalake);
     } catch (IOException e) {
       throw new RuntimeException("Failed to register the job entity " + jobEntity, e);
     }
@@ -669,6 +692,14 @@ void pullAndUpdateJobStatus() {
                                 e);
                           }
                         });
+              } catch (OptimisticLockException e) {
+                // A later poll re-reads both executor state and metadata. Never stop the scheduled
+                // task or replay external submission/cancellation because a metadata CAS lost.
+                LOG.info(
+                    "Job {} under metalake {} changed concurrently; deferring status update",
+                    job.name(),
+                    metalake);
+                return;
               } catch (NoSuchEntityException e) {
                 // The job could have been deleted concurrently (e.g. by legacy-timeline cleanup)
                 // in the gap between the listJobs() snapshot above and this update. Skip it rather
@@ -767,11 +798,7 @@ void cleanUpStagingDirs() {
     for (String metalake : metalakes) {
       List<JobEntity> finishedJobs =
           listJobs(metalake, Optional.empty()).stream()
-              .filter(
-                  job ->
-                      job.status() == JobHandle.Status.CANCELLED
-                          || job.status() == JobHandle.Status.SUCCEEDED
-                          || job.status() == JobHandle.Status.FAILED)
+              .filter(job -> isFinishedStatus(job.status()))
               .filter(
                   job ->
                       job.finishedAt() > 0
@@ -793,6 +820,13 @@ void cleanUpStagingDirs() {
                 FileUtils.deleteDirectory(jobStagingDir);
                 LOG.info("Deleted job staging directory {} for job {}", jobStagingPath, job.name());
               }
+            } catch (OptimisticLockException e) {
+              // Keep the files when deletion loses its CAS. The next cleanup run re-reads the
+              // job and checks retention eligibility again; this batch can process other jobs.
+              LOG.info(
+                  "Job {} under metalake {} changed concurrently; deferring cleanup",
+                  job.name(),
+                  metalake);
             } catch (IOException e) {
               LOG.error("Failed to delete job and staging directory for job {}", job.name(), e);
             }

core/src/main/java/org/apache/gravitino/storage/relational/mapper/JobMetaMapper.java

@@ -19,6 +19,7 @@
 package org.apache.gravitino.storage.relational.mapper;
 
 import java.util.List;
+import javax.annotation.Nullable;
 import org.apache.gravitino.storage.relational.po.JobPO;
 import org.apache.ibatis.annotations.DeleteProvider;
 import org.apache.ibatis.annotations.InsertProvider;
@@ -53,12 +54,6 @@ JobPO selectJobPOByMetalakeAndRunId(
   @UpdateProvider(type = JobMetaSQLProviderFactory.class, method = "updateJobMeta")
   Integer updateJobMeta(@Param("newJobMeta") JobPO newJobPO, @Param("oldJobMeta") JobPO oldJobPO);
 
-  @UpdateProvider(
-      type = JobMetaSQLProviderFactory.class,
-      method = "softDeleteJobMetaByMetalakeAndTemplate")
-  Integer softDeleteJobMetaByMetalakeAndTemplate(
-      @Param("metalakeName") String metalakeName, @Param("jobTemplateName") String jobTemplateName);
-
   @UpdateProvider(type = JobMetaSQLProviderFactory.class, method = "softDeleteJobMetasByMetalakeId")
   void softDeleteJobMetasByMetalakeId(@Param("metalakeId") Long metalakeId);
 
@@ -71,10 +66,49 @@ Integer softDeleteJobMetaByMetalakeAndTemplate(
   Integer deleteJobMetasByLegacyTimeline(
       @Param("legacyTimeline") Long legacyTimeline, @Param("limit") int limit);
 
-  @UpdateProvider(type = JobMetaSQLProviderFactory.class, method = "softDeleteJobMetaByRunId")
-  Integer softDeleteJobMetaByRunId(@Param("jobRunId") Long jobRunId);
-
   @SelectProvider(type = JobMetaSQLProviderFactory.class, method = "batchSelectJobByRunIds")
   List<JobPO> batchSelectJobByRunIds(
       @Param("metalakeName") String metalakeName, @Param("jobRunIds") List<Long> jobRunIds);
+  /**
+   * Locks the active row for OCC identity validation.
+   *
+   * @param jobRunId the stable job run ID
+   * @param metalakeId the owning metalake ID
+   * @return the active row identity, or null if missing
+   */
+  @Nullable
+  @SelectProvider(type = JobMetaSQLProviderFactory.class, method = "selectJobRunIdForUpdate")
+  Long selectJobRunIdForUpdate(
+      @Param("jobRunId") Long jobRunId, @Param("metalakeId") Long metalakeId);
+
+  /**
+   * Deletes active metadata using a stable identity and expected version.
+   *
+   * @param jobRunId the stable job run ID
+   * @param currentVersion the expected OCC version
+   * @return the affected row count
+   */
+  @UpdateProvider(
+      type = JobMetaSQLProviderFactory.class,
+      method = "softDeleteJobByRunIdWithVersion")
+  int softDeleteJobByRunIdWithVersion(
+      @Param("jobRunId") Long jobRunId, @Param("currentVersion") Long currentVersion);
+
+  /**
+   * Deletes active metadata using a stable identity.
+   *
+   * @param jobTemplateId the stable template ID
+   * @return the affected row count
+   */
+  @UpdateProvider(type = JobMetaSQLProviderFactory.class, method = "softDeleteJobsByTemplateId")
+  int softDeleteJobsByTemplateId(@Param("jobTemplateId") Long jobTemplateId);
+  /**
+   * Locks a nonterminal job belonging to the template using a current database read.
+   *
+   * @param jobTemplateId the stable template ID
+   * @return a nonterminal job ID, or null if there are none
+   */
+  @Nullable
+  @SelectProvider(type = JobMetaSQLProviderFactory.class, method = "selectNonterminalJobForUpdate")
+  Long selectNonterminalJobForUpdate(@Param("jobTemplateId") Long jobTemplateId);
 }

core/src/main/java/org/apache/gravitino/storage/relational/mapper/JobMetaSQLProviderFactory.java

@@ -81,12 +81,6 @@ public static String updateJobMeta(
     return getProvider().updateJobMeta(newJobPO, oldJobPO);
   }
 
-  public static String softDeleteJobMetaByMetalakeAndTemplate(
-      @Param("metalakeName") String metalakeName,
-      @Param("jobTemplateName") String jobTemplateName) {
-    return getProvider().softDeleteJobMetaByMetalakeAndTemplate(metalakeName, jobTemplateName);
-  }
-
   public static String softDeleteJobMetasByMetalakeId(@Param("metalakeId") Long metalakeId) {
     return getProvider().softDeleteJobMetasByMetalakeId(metalakeId);
   }
@@ -101,12 +95,52 @@ public static String softDeleteJobMetasByLegacyTimeline(
     return getProvider().softDeleteJobMetasByLegacyTimeline(legacyTimeline);
   }
 
-  public static String softDeleteJobMetaByRunId(@Param("jobRunId") Long jobRunId) {
-    return getProvider().softDeleteJobMetaByRunId(jobRunId);
-  }
-
   public static String batchSelectJobByRunIds(
       @Param("metalakeName") String metalakeName, @Param("jobRunIds") List<Long> jobRunIds) {
     return getProvider().batchSelectJobByRunIds(metalakeName, jobRunIds);
   }
+
+  /**
+   * Locks the active row for OCC identity validation.
+   *
+   * @param jobRunId the stable job run ID
+   * @param metalakeId the owning metalake ID
+   * @return the SQL statement
+   */
+  public static String selectJobRunIdForUpdate(
+      @Param("jobRunId") Long jobRunId, @Param("metalakeId") Long metalakeId) {
+    return getProvider().selectJobRunIdForUpdate(jobRunId, metalakeId);
+  }
+
+  /**
+   * Deletes active metadata using a stable identity and expected version.
+   *
+   * @param jobRunId the stable job run ID
+   * @param currentVersion the expected OCC version
+   * @return the SQL statement
+   */
+  public static String softDeleteJobByRunIdWithVersion(
+      @Param("jobRunId") Long jobRunId, @Param("currentVersion") Long currentVersion) {
+    return getProvider().softDeleteJobByRunIdWithVersion(jobRunId, currentVersion);
+  }
+
+  /**
+   * Deletes active metadata using a stable identity.
+   *
+   * @param jobTemplateId the stable template ID
+   * @return the SQL statement
+   */
+  public static String softDeleteJobsByTemplateId(@Param("jobTemplateId") Long jobTemplateId) {
+    return getProvider().softDeleteJobsByTemplateId(jobTemplateId);
+  }
+
+  /**
+   * Builds a locking lookup for a nonterminal job belonging to a template.
+   *
+   * @param jobTemplateId the stable template ID
+   * @return the SQL statement
+   */
+  public static String selectNonterminalJobForUpdate(@Param("jobTemplateId") Long jobTemplateId) {
+    return getProvider().selectNonterminalJobForUpdate(jobTemplateId);
+  }
 }

core/src/main/java/org/apache/gravitino/storage/relational/mapper/JobTemplateMetaMapper.java

@@ -19,6 +19,7 @@
 package org.apache.gravitino.storage.relational.mapper;
 
 import java.util.List;
+import javax.annotation.Nullable;
 import org.apache.gravitino.storage.relational.po.JobTemplatePO;
 import org.apache.ibatis.annotations.DeleteProvider;
 import org.apache.ibatis.annotations.InsertProvider;
@@ -49,12 +50,6 @@ void insertJobTemplateMetaOnDuplicateKeyUpdate(
   JobTemplatePO selectJobTemplatePOByMetalakeAndName(
       @Param("metalakeName") String metalakeName, @Param("jobTemplateName") String jobTemplateName);
 
-  @UpdateProvider(
-      type = JobTemplateMetaSQLProviderFactory.class,
-      method = "softDeleteJobTemplateMetaByMetalakeAndName")
-  Integer softDeleteJobTemplateMetaByMetalakeAndName(
-      @Param("metalakeName") String metalakeName, @Param("jobTemplateName") String jobTemplateName);
-
   @UpdateProvider(
       type = JobTemplateMetaSQLProviderFactory.class,
       method = "softDeleteJobTemplateMetasByMetalakeId")
@@ -92,4 +87,40 @@ List<JobTemplatePO> listJobTemplatePOsByJobTemplateIds(
   List<JobTemplatePO> batchSelectJobTemplateByIdentifier(
       @Param("metalakeName") String metalakeName,
       @Param("jobTemplateNames") List<String> jobTemplateNames);
+  /**
+   * Locks the active row for OCC identity validation.
+   *
+   * @param jobTemplateId the stable template ID
+   * @return the active row identity, or null if missing
+   */
+  @Nullable
+  @SelectProvider(
+      type = JobTemplateMetaSQLProviderFactory.class,
+      method = "selectJobTemplateByIdForUpdate")
+  JobTemplatePO selectJobTemplateByIdForUpdate(@Param("jobTemplateId") Long jobTemplateId);
+
+  /**
+   * Locks the active row for OCC identity validation.
+   *
+   * @param jobTemplateId the stable template ID
+   * @return the active row identity, or null if missing
+   */
+  @Nullable
+  @SelectProvider(
+      type = JobTemplateMetaSQLProviderFactory.class,
+      method = "selectJobTemplateByIdForShare")
+  JobTemplatePO selectJobTemplateByIdForShare(@Param("jobTemplateId") Long jobTemplateId);
+
+  /**
+   * Deletes active metadata using a stable identity and expected version.
+   *
+   * @param jobTemplateId the stable template ID
+   * @param currentVersion the expected OCC version
+   * @return the affected row count
+   */
+  @UpdateProvider(
+      type = JobTemplateMetaSQLProviderFactory.class,
+      method = "softDeleteJobTemplateById")
+  int softDeleteJobTemplateById(
+      @Param("jobTemplateId") Long jobTemplateId, @Param("currentVersion") Long currentVersion);
 }