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

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

@@ -57,6 +57,8 @@
 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.OptimisticLockException;
 import org.apache.gravitino.json.JsonUtils;
 import org.apache.gravitino.lock.LockType;
 import org.apache.gravitino.lock.TreeLockUtils;
@@ -226,6 +228,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);
           }
@@ -280,6 +284,24 @@ public boolean deleteJobTemplate(String metalake, String jobTemplateName) throws
           jobTemplateName, metalake);
     }
 
+    // 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 (IOException ioe) {
+                throw new RuntimeException(ioe);
+              }
+            });
+    if (!deleted) {
+      return false;
+    }
+
     // Delete all the job staging directories associated with the job template.
     String jobTemplateStagingPath =
         stagingDir.getAbsolutePath() + File.separator + metalake + File.separator + jobTemplateName;
@@ -292,19 +314,7 @@ public boolean deleteJobTemplate(String metalake, String jobTemplateName) throws
       }
     }
 
-    // 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 +345,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 +498,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 +691,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
@@ -793,6 +823,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,40 @@ 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);
 }

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,41 @@ 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);
+  }
 }

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);
 }

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

@@ -51,7 +51,12 @@ public static JobTemplateMetaBaseSQLProvider getProvider() {
 
   static class JobTemplateMetaMySQLProvider extends JobTemplateMetaBaseSQLProvider {}
 
-  static class JobTemplateMetaH2Provider extends JobTemplateMetaBaseSQLProvider {}
+  static class JobTemplateMetaH2Provider extends JobTemplateMetaBaseSQLProvider {
+    @Override
+    public String selectJobTemplateByIdForShare(Long jobTemplateId) {
+      return selectJobTemplateByIdForUpdate(jobTemplateId);
+    }
+  }
 
   public static String insertJobTemplateMeta(
       @Param("jobTemplateMeta") JobTemplatePO jobTemplatePO) {
@@ -73,12 +78,6 @@ public static String selectJobTemplatePOByMetalakeAndName(
     return getProvider().selectJobTemplatePOByMetalakeAndName(metalakeName, jobTemplateName);
   }
 
-  public static String softDeleteJobTemplateMetaByMetalakeAndName(
-      @Param("metalakeName") String metalakeName,
-      @Param("jobTemplateName") String jobTemplateName) {
-    return getProvider().softDeleteJobTemplateMetaByMetalakeAndName(metalakeName, jobTemplateName);
-  }
-
   public static String softDeleteJobTemplateMetasByMetalakeId(
       @Param("metalakeId") Long metalakeId) {
     return getProvider().softDeleteJobTemplateMetasByMetalakeId(metalakeId);
@@ -114,4 +113,35 @@ public static String batchSelectJobTemplateByIdentifier(
       @Param("jobTemplateNames") List<String> jobTemplateNames) {
     return getProvider().batchSelectJobTemplateByIdentifier(metalakeName, jobTemplateNames);
   }
+  /**
+   * Locks the active row for OCC identity validation.
+   *
+   * @param jobTemplateId the stable template ID
+   * @return the SQL statement
+   */
+  public static String selectJobTemplateByIdForUpdate(@Param("jobTemplateId") Long jobTemplateId) {
+    return getProvider().selectJobTemplateByIdForUpdate(jobTemplateId);
+  }
+
+  /**
+   * Locks the active row for OCC identity validation.
+   *
+   * @param jobTemplateId the stable template ID
+   * @return the SQL statement
+   */
+  public static String selectJobTemplateByIdForShare(@Param("jobTemplateId") Long jobTemplateId) {
+    return getProvider().selectJobTemplateByIdForShare(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 SQL statement
+   */
+  public static String softDeleteJobTemplateById(
+      @Param("jobTemplateId") Long jobTemplateId, @Param("currentVersion") Long currentVersion) {
+    return getProvider().softDeleteJobTemplateById(jobTemplateId, currentVersion);
+  }
 }