diff --git a/core/src/main/java/org/apache/gravitino/job/JobManager.java b/core/src/main/java/org/apache/gravitino/job/JobManager.java index 6db68afb14d..e8365cda86d 100644 --- a/core/src/main/java/org/apache/gravitino/job/JobManager.java +++ b/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 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); } diff --git a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/JobMetaMapper.java b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/JobMetaMapper.java index ae667fcad16..c11d1a92e37 100644 --- a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/JobMetaMapper.java +++ b/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 batchSelectJobByRunIds( @Param("metalakeName") String metalakeName, @Param("jobRunIds") List 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); } diff --git a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/JobMetaSQLProviderFactory.java b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/JobMetaSQLProviderFactory.java index 20d6c414e17..f75ce6e18f4 100644 --- a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/JobMetaSQLProviderFactory.java +++ b/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 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); + } } diff --git a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/JobTemplateMetaMapper.java b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/JobTemplateMetaMapper.java index 09f5d52f629..d17521b2c88 100644 --- a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/JobTemplateMetaMapper.java +++ b/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 listJobTemplatePOsByJobTemplateIds( List batchSelectJobTemplateByIdentifier( @Param("metalakeName") String metalakeName, @Param("jobTemplateNames") List 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); } diff --git a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/JobTemplateMetaSQLProviderFactory.java b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/JobTemplateMetaSQLProviderFactory.java index 04383da401e..8e59c1d0b4b 100644 --- a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/JobTemplateMetaSQLProviderFactory.java +++ b/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 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); + } } diff --git a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/JobMetaBaseSQLProvider.java b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/JobMetaBaseSQLProvider.java index 344f4007fdc..89ffd79aedf 100644 --- a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/JobMetaBaseSQLProvider.java +++ b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/JobMetaBaseSQLProvider.java @@ -159,25 +159,6 @@ public String updateJobMeta( + " last_version = #{newJobMeta.lastVersion}" + " WHERE job_run_id = #{oldJobMeta.jobRunId}" + " AND current_version = #{oldJobMeta.currentVersion}" - + " AND last_version = #{oldJobMeta.lastVersion}" - + " AND deleted_at = 0"; - } - - public String softDeleteJobMetaByMetalakeAndTemplate( - @Param("metalakeName") String metalakeName, - @Param("jobTemplateName") String jobTemplateName) { - return "UPDATE " - + JobMetaMapper.TABLE_NAME - + " SET deleted_at = " - + DatabaseTimeSQL.MYSQL - + " WHERE metalake_id = (" - + " SELECT metalake_id FROM " - + MetalakeMetaMapper.TABLE_NAME - + " WHERE metalake_name = #{metalakeName} AND deleted_at = 0)" - + " AND job_template_id IN (" - + " SELECT job_template_id FROM " - + JobTemplateMetaMapper.TABLE_NAME - + " WHERE job_template_name = #{jobTemplateName} AND deleted_at = 0)" + " AND deleted_at = 0"; } @@ -189,14 +170,6 @@ public String softDeleteJobMetasByMetalakeId(@Param("metalakeId") Long metalakeI + " WHERE metalake_id = #{metalakeId} AND deleted_at = 0"; } - public String softDeleteJobMetaByRunId(@Param("jobRunId") Long jobRunId) { - return "UPDATE " - + JobMetaMapper.TABLE_NAME - + " SET deleted_at = " - + DatabaseTimeSQL.MYSQL - + " WHERE job_run_id = #{jobRunId} AND deleted_at = 0"; - } - public String softDeleteJobMetasByLegacyTimeline(@Param("legacyTimeline") Long legacyTimeline) { return "UPDATE " + JobMetaMapper.TABLE_NAME @@ -241,4 +214,61 @@ public String batchSelectJobByRunIds( + " AND jrm.deleted_at = 0 AND jtm.deleted_at = 0 AND mm.deleted_at = 0" + ""; } + + /** + * 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 String selectJobRunIdForUpdate( + @Param("jobRunId") Long jobRunId, @Param("metalakeId") Long metalakeId) { + return "SELECT job_run_id FROM " + + JobMetaMapper.TABLE_NAME + + " WHERE job_run_id = #{jobRunId} AND metalake_id = #{metalakeId} AND deleted_at = 0 FOR UPDATE"; + } + + /** + * 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 String softDeleteJobByRunIdWithVersion( + @Param("jobRunId") Long jobRunId, @Param("currentVersion") Long currentVersion) { + return "UPDATE " + + JobMetaMapper.TABLE_NAME + + " SET deleted_at = " + + DatabaseTimeSQL.MYSQL + + " WHERE job_run_id = #{jobRunId} AND current_version = #{currentVersion} AND deleted_at = 0"; + } + + /** + * Deletes active metadata using a stable identity. + * + * @param jobTemplateId the stable template ID + * @return the SQL statement + */ + public String softDeleteJobsByTemplateId(@Param("jobTemplateId") Long jobTemplateId) { + return "UPDATE " + + JobMetaMapper.TABLE_NAME + + " SET deleted_at = " + + DatabaseTimeSQL.MYSQL + + " WHERE job_template_id = #{jobTemplateId} AND deleted_at = 0"; + } + + /** + * Builds a current read that locks a nonterminal job without joining the deleted template. + * + * @param jobTemplateId the stable template ID + * @return the SQL statement + */ + public String selectNonterminalJobForUpdate(@Param("jobTemplateId") Long jobTemplateId) { + return "SELECT job_run_id FROM " + + JobMetaMapper.TABLE_NAME + + " WHERE job_template_id = #{jobTemplateId} AND deleted_at = 0" + + " AND job_run_status NOT IN ('SUCCEEDED', 'FAILED', 'CANCELLED') LIMIT 1 FOR UPDATE"; + } } diff --git a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/JobTemplateMetaBaseSQLProvider.java b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/JobTemplateMetaBaseSQLProvider.java index 747e98e6e07..002c57b6470 100644 --- a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/JobTemplateMetaBaseSQLProvider.java +++ b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/JobTemplateMetaBaseSQLProvider.java @@ -94,20 +94,6 @@ public String selectJobTemplatePOByMetalakeAndName( + " AND jtm.deleted_at = 0 AND mm.deleted_at = 0"; } - public String softDeleteJobTemplateMetaByMetalakeAndName( - @Param("metalakeName") String metalakeName, - @Param("jobTemplateName") String jobTemplateName) { - return "UPDATE " - + JobTemplateMetaMapper.TABLE_NAME - + " SET deleted_at = " - + DatabaseTimeSQL.MYSQL - + " WHERE job_template_name = #{jobTemplateName} AND metalake_id =" - + " (SELECT metalake_id FROM " - + MetalakeMetaMapper.TABLE_NAME - + " WHERE metalake_name = #{metalakeName} AND deleted_at = 0)" - + " AND deleted_at = 0"; - } - public String softDeleteJobTemplateMetasByMetalakeId(@Param("metalakeId") Long metalakeId) { return "UPDATE " + JobTemplateMetaMapper.TABLE_NAME @@ -137,10 +123,7 @@ public String updateJobTemplateMeta( + " last_version = #{newJobTemplateMeta.lastVersion}," + " deleted_at = #{newJobTemplateMeta.deletedAt}" + " WHERE job_template_id = #{oldJobTemplateMeta.jobTemplateId}" - + " AND job_template_name = #{oldJobTemplateMeta.jobTemplateName}" - + " AND metalake_id = #{oldJobTemplateMeta.metalakeId}" + " AND current_version = #{oldJobTemplateMeta.currentVersion}" - + " AND last_version = #{oldJobTemplateMeta.lastVersion}" + " AND deleted_at = 0"; } @@ -208,4 +191,52 @@ public String batchSelectJobTemplateByIdentifier( + " AND jtm.deleted_at = 0 AND mm.deleted_at = 0" + ""; } + + /** + * Locks the active row for OCC identity validation. + * + * @param jobTemplateId the stable template ID + * @return the SQL statement + */ + public String selectJobTemplateByIdForUpdate(@Param("jobTemplateId") Long jobTemplateId) { + return selectJobTemplateIdentityById() + " FOR UPDATE"; + } + + /** + * Locks the active row for OCC identity validation. + * + * @param jobTemplateId the stable template ID + * @return the SQL statement + */ + public String selectJobTemplateByIdForShare(@Param("jobTemplateId") Long jobTemplateId) { + return selectJobTemplateIdentityById() + " LOCK IN SHARE MODE"; + } + + /** + * 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 String softDeleteJobTemplateById( + @Param("jobTemplateId") Long jobTemplateId, @Param("currentVersion") Long currentVersion) { + return "UPDATE " + + JobTemplateMetaMapper.TABLE_NAME + + " SET deleted_at = " + + DatabaseTimeSQL.MYSQL + + " WHERE job_template_id = #{jobTemplateId} AND current_version = #{currentVersion} AND deleted_at = 0"; + } + + /** + * Builds the identity projection used by locking reads. + * + * @return SQL selecting the active template's identity fields + */ + protected String selectJobTemplateIdentityById() { + return "SELECT job_template_id AS jobTemplateId, job_template_name AS jobTemplateName," + + " metalake_id AS metalakeId FROM " + + JobTemplateMetaMapper.TABLE_NAME + + " WHERE job_template_id = #{jobTemplateId} AND deleted_at = 0"; + } } diff --git a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/JobMetaPostgreSQLProvider.java b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/JobMetaPostgreSQLProvider.java index f2169ca4e99..1e73489f094 100644 --- a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/JobMetaPostgreSQLProvider.java +++ b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/JobMetaPostgreSQLProvider.java @@ -20,7 +20,6 @@ import org.apache.gravitino.storage.relational.mapper.JobMetaMapper; import org.apache.gravitino.storage.relational.mapper.JobTemplateMetaMapper; -import org.apache.gravitino.storage.relational.mapper.MetalakeMetaMapper; import org.apache.gravitino.storage.relational.mapper.provider.DatabaseTimeSQL; import org.apache.gravitino.storage.relational.mapper.provider.base.JobMetaBaseSQLProvider; import org.apache.gravitino.storage.relational.po.JobPO; @@ -62,25 +61,6 @@ public String insertJobMetaOnDuplicateKeyUpdate(@Param("jobMeta") JobPO jobPO) { + " deleted_at = #{jobMeta.deletedAt}"; } - @Override - public String softDeleteJobMetaByMetalakeAndTemplate( - @Param("metalakeName") String metalakeName, - @Param("jobTemplateName") String jobTemplateName) { - return "UPDATE " - + JobMetaMapper.TABLE_NAME - + " SET deleted_at = " - + DatabaseTimeSQL.POSTGRESQL - + " WHERE metalake_id IN (" - + " SELECT metalake_id FROM " - + MetalakeMetaMapper.TABLE_NAME - + " WHERE metalake_name = #{metalakeName} AND deleted_at = 0)" - + " AND job_template_id IN (" - + " SELECT job_template_id FROM " - + JobTemplateMetaMapper.TABLE_NAME - + " WHERE job_template_name = #{jobTemplateName} AND deleted_at = 0)" - + " AND deleted_at = 0"; - } - @Override public String softDeleteJobMetasByMetalakeId(@Param("metalakeId") Long metalakeId) { return "UPDATE " @@ -90,14 +70,6 @@ public String softDeleteJobMetasByMetalakeId(@Param("metalakeId") Long metalakeI + " WHERE metalake_id = #{metalakeId} AND deleted_at = 0"; } - public String softDeleteJobMetaByRunId(@Param("jobRunId") Long jobRunId) { - return "UPDATE " - + JobMetaMapper.TABLE_NAME - + " SET deleted_at = " - + DatabaseTimeSQL.POSTGRESQL - + " WHERE job_run_id = #{jobRunId} AND deleted_at = 0"; - } - @Override public String softDeleteJobMetasByLegacyTimeline(@Param("legacyTimeline") Long legacyTimeline) { return "UPDATE " @@ -116,4 +88,23 @@ public String deleteJobMetasByLegacyTimeline( + JobMetaMapper.TABLE_NAME + " WHERE deleted_at < #{legacyTimeline} AND deleted_at > 0 LIMIT #{limit})"; } + + @Override + public String softDeleteJobByRunIdWithVersion( + @Param("jobRunId") Long jobRunId, @Param("currentVersion") Long currentVersion) { + return "UPDATE " + + JobMetaMapper.TABLE_NAME + + " SET deleted_at = " + + DatabaseTimeSQL.POSTGRESQL + + " WHERE job_run_id = #{jobRunId} AND current_version = #{currentVersion} AND deleted_at = 0"; + } + + @Override + public String softDeleteJobsByTemplateId(@Param("jobTemplateId") Long jobTemplateId) { + return "UPDATE " + + JobMetaMapper.TABLE_NAME + + " SET deleted_at = " + + DatabaseTimeSQL.POSTGRESQL + + " WHERE job_template_id = #{jobTemplateId} AND deleted_at = 0"; + } } diff --git a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/JobTemplateMetaPostgreSQLProvider.java b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/JobTemplateMetaPostgreSQLProvider.java index e7426ff45b2..50db19ca31a 100644 --- a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/JobTemplateMetaPostgreSQLProvider.java +++ b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/JobTemplateMetaPostgreSQLProvider.java @@ -19,7 +19,6 @@ package org.apache.gravitino.storage.relational.mapper.provider.postgresql; import org.apache.gravitino.storage.relational.mapper.JobTemplateMetaMapper; -import org.apache.gravitino.storage.relational.mapper.MetalakeMetaMapper; import org.apache.gravitino.storage.relational.mapper.provider.DatabaseTimeSQL; import org.apache.gravitino.storage.relational.mapper.provider.base.JobTemplateMetaBaseSQLProvider; import org.apache.gravitino.storage.relational.po.JobTemplatePO; @@ -27,21 +26,6 @@ public class JobTemplateMetaPostgreSQLProvider extends JobTemplateMetaBaseSQLProvider { - @Override - public String softDeleteJobTemplateMetaByMetalakeAndName( - @Param("metalakeName") String metalakeName, - @Param("jobTemplateName") String jobTemplateName) { - return "UPDATE " - + JobTemplateMetaMapper.TABLE_NAME - + " SET deleted_at = " - + DatabaseTimeSQL.POSTGRESQL - + " WHERE metalake_id IN (" - + " SELECT metalake_id FROM " - + MetalakeMetaMapper.TABLE_NAME - + " WHERE metalake_name = #{metalakeName} AND deleted_at = 0)" - + " AND job_template_name = #{jobTemplateName} AND deleted_at = 0"; - } - @Override public String softDeleteJobTemplateMetasByMetalakeId(@Param("metalakeId") Long metalakeId) { return "UPDATE " @@ -82,4 +66,19 @@ public String deleteJobTemplateMetasByLegacyTimeline(Long legacyTimeline, int li + JobTemplateMetaMapper.TABLE_NAME + " WHERE deleted_at < #{legacyTimeline} AND deleted_at > 0 LIMIT #{limit})"; } + + @Override + public String softDeleteJobTemplateById( + @Param("jobTemplateId") Long jobTemplateId, @Param("currentVersion") Long currentVersion) { + return "UPDATE " + + JobTemplateMetaMapper.TABLE_NAME + + " SET deleted_at = " + + DatabaseTimeSQL.POSTGRESQL + + " WHERE job_template_id = #{jobTemplateId} AND current_version = #{currentVersion} AND deleted_at = 0"; + } + + @Override + public String selectJobTemplateByIdForShare(Long jobTemplateId) { + return selectJobTemplateIdentityById() + " FOR SHARE"; + } } diff --git a/core/src/main/java/org/apache/gravitino/storage/relational/service/JobMetaService.java b/core/src/main/java/org/apache/gravitino/storage/relational/service/JobMetaService.java index a67b8a21f3b..a07a13ce0f4 100644 --- a/core/src/main/java/org/apache/gravitino/storage/relational/service/JobMetaService.java +++ b/core/src/main/java/org/apache/gravitino/storage/relational/service/JobMetaService.java @@ -103,17 +103,28 @@ public void insertJob(JobEntity jobEntity, boolean overwrite) throws IOException JobPO.JobPOBuilder builder = JobPO.builder().withMetalakeId(metalakeId); JobPO jobPO = JobPO.initializeJobPO(jobEntity, builder); - SessionUtils.doWithCommit( - JobMetaMapper.class, - mapper -> { - if (overwrite) { - mapper.insertJobMetaOnDuplicateKeyUpdate(jobPO); - } else { - mapper.insertJobMeta(jobPO); - } - }); + long templateId = + JobTemplateMetaService.getInstance() + .getJobTemplateIdByMetalakeIdAndName(metalakeId, jobEntity.jobTemplateName()); + SessionUtils.doMultipleWithCommit( + () -> + MetalakeMetaService.getInstance().lockMetalakeForChildWrite(metalakeName, metalakeId), + () -> + JobTemplateMetaService.getInstance() + .lockTemplateForJobWrite(jobEntity.jobTemplateName(), templateId, metalakeId), + () -> + SessionUtils.doWithoutCommit( + JobMetaMapper.class, + mapper -> { + if (overwrite) { + mapper.insertJobMetaOnDuplicateKeyUpdate(jobPO); + } else { + mapper.insertJobMeta(jobPO); + } + })); } catch (RuntimeException e) { ExceptionUtils.checkSQLException(e, Entity.EntityType.JOB, jobEntity.id().toString()); + throw e; } } @@ -132,38 +143,31 @@ public JobEntity updateJob( JobPO.JobPOBuilder newBuilder = JobPO.builder().withMetalakeId(oldJobPO.metalakeId()); JobPO newJobPO = JobPO.updateJobPO(oldJobPO, newJobEntity, newBuilder); - Integer result; try { - result = - SessionUtils.doWithCommitAndFetchResult( - JobMetaMapper.class, mapper -> mapper.updateJobMeta(newJobPO, oldJobPO)); + SessionUtils.doMultipleWithCommit( + () -> + OccWriteSupport.updateWithVersion( + () -> + SessionUtils.getWithoutCommit( + JobMetaMapper.class, mapper -> mapper.updateJobMeta(newJobPO, oldJobPO)), + () -> writeFailure(jobIdent, oldJobPO))); } catch (RuntimeException e) { - ExceptionUtils.checkSQLException(e, Entity.EntityType.JOB, oldJobEntity.name()); + ExceptionUtils.checkSQLException(e, Entity.EntityType.JOB, jobIdent.name()); throw e; } - - if (result == null || result == 0) { - throw new NoSuchEntityException( - NoSuchEntityException.NO_SUCH_ENTITY_MESSAGE, - Entity.EntityType.JOB.name().toLowerCase(Locale.ROOT), - oldJobEntity.name()); - } else if (result > 1) { - throw new IOException( - String.format( - "Failed to update job: %s, because more than one rows are updated: %d", - oldJobEntity.name(), result)); - } else { - return newJobEntity; - } + return newJobEntity; } @Monitored(metricsSource = GRAVITINO_RELATIONAL_STORE_METRIC_NAME, baseMetricName = "deleteJob") public boolean deleteJob(NameIdentifier jobIdent) { - long jobRunIdLong = parseJobRunId(jobIdent.name()); - int result = - SessionUtils.doWithCommitAndFetchResult( - JobMetaMapper.class, mapper -> mapper.softDeleteJobMetaByRunId(jobRunIdLong)); - return result > 0; + // Preserve malformed-ID validation even for an otherwise missing job. + parseJobRunId(jobIdent.name()); + try { + deleteJobWithVersion(jobIdent, getJobPO(jobIdent)); + return true; + } catch (NoSuchEntityException e) { + return false; + } } @Monitored( @@ -240,4 +244,32 @@ public List batchGetJobByIdentifier(List identifiers) .collect(Collectors.toList()); }); } + + /** Deletes a job only if the observed version is still active. */ + void deleteJobWithVersion(NameIdentifier ident, JobPO observed) { + SessionUtils.doMultipleWithCommit( + () -> + OccWriteSupport.deleteWithVersion( + () -> + SessionUtils.getWithoutCommit( + JobMetaMapper.class, + mapper -> + mapper.softDeleteJobByRunIdWithVersion( + observed.jobRunId(), observed.currentVersion())), + () -> writeFailure(ident, observed))); + } + + private RuntimeException writeFailure(NameIdentifier ident, JobPO observed) { + // Read only the job row: joining and locking the template here would invert cascade lock order. + return OccWriteSupport.writeFailure( + ident, + Entity.EntityType.JOB, + () -> + SessionUtils.getWithoutCommit( + JobMetaMapper.class, + mapper -> + mapper.selectJobRunIdForUpdate(observed.jobRunId(), observed.metalakeId())), + null, + null); + } } diff --git a/core/src/main/java/org/apache/gravitino/storage/relational/service/JobTemplateMetaService.java b/core/src/main/java/org/apache/gravitino/storage/relational/service/JobTemplateMetaService.java index 15685af4928..9f5dffbfa8e 100644 --- a/core/src/main/java/org/apache/gravitino/storage/relational/service/JobTemplateMetaService.java +++ b/core/src/main/java/org/apache/gravitino/storage/relational/service/JobTemplateMetaService.java @@ -25,7 +25,6 @@ import java.util.List; import java.util.Locale; import java.util.Objects; -import java.util.concurrent.atomic.AtomicInteger; import java.util.function.Function; import java.util.stream.Collectors; import org.apache.gravitino.Entity; @@ -33,6 +32,7 @@ import org.apache.gravitino.NameIdentifier; import org.apache.gravitino.Namespace; import org.apache.gravitino.exceptions.NoSuchEntityException; +import org.apache.gravitino.exceptions.NonEmptyEntityException; import org.apache.gravitino.meta.JobTemplateEntity; import org.apache.gravitino.metrics.Monitored; import org.apache.gravitino.storage.relational.mapper.JobMetaMapper; @@ -91,15 +91,19 @@ public void insertJobTemplate(JobTemplateEntity jobTemplateEntity, boolean overw JobTemplatePO jobTemplatePO = JobTemplatePO.initializeJobTemplatePO(jobTemplateEntity, builder); - SessionUtils.doWithCommit( - JobTemplateMetaMapper.class, - mapper -> { - if (overwrite) { - mapper.insertJobTemplateMetaOnDuplicateKeyUpdate(jobTemplatePO); - } else { - mapper.insertJobTemplateMeta(jobTemplatePO); - } - }); + SessionUtils.doMultipleWithCommit( + () -> + MetalakeMetaService.getInstance().lockMetalakeForChildWrite(metalakeName, metalakeId), + () -> + SessionUtils.doWithoutCommit( + JobTemplateMetaMapper.class, + mapper -> { + if (overwrite) { + mapper.insertJobTemplateMetaOnDuplicateKeyUpdate(jobTemplatePO); + } else { + mapper.insertJobTemplateMeta(jobTemplatePO); + } + })); } catch (RuntimeException e) { ExceptionUtils.checkSQLException(e, Entity.EntityType.JOB_TEMPLATE, jobTemplateEntity.name()); throw e; @@ -110,24 +114,12 @@ public void insertJobTemplate(JobTemplateEntity jobTemplateEntity, boolean overw metricsSource = GRAVITINO_RELATIONAL_STORE_METRIC_NAME, baseMetricName = "deleteJobTemplate") public boolean deleteJobTemplate(NameIdentifier jobTemplateIdent) { - String metalakeName = jobTemplateIdent.namespace().level(0); - String jobTemplateName = jobTemplateIdent.name(); - - AtomicInteger result = new AtomicInteger(0); - SessionUtils.doMultipleWithCommit( - () -> - SessionUtils.doWithoutCommit( - JobMetaMapper.class, - mapper -> - mapper.softDeleteJobMetaByMetalakeAndTemplate(metalakeName, jobTemplateName)), - () -> - result.set( - SessionUtils.getWithoutCommit( - JobTemplateMetaMapper.class, - mapper -> - mapper.softDeleteJobTemplateMetaByMetalakeAndName( - metalakeName, jobTemplateName)))); - return result.get() > 0; + try { + deleteJobTemplateWithVersion(jobTemplateIdent, getJobTemplatePO(jobTemplateIdent)); + return true; + } catch (NoSuchEntityException e) { + return false; + } } @Monitored( @@ -160,31 +152,21 @@ public JobTemplateEntity updateJobTemplate( JobTemplatePO newJobTemplatePO = JobTemplatePO.updateJobTemplatePO(oldJobTemplatePO, newJobTemplateEntity, newBuilder); - Integer result; try { - result = - SessionUtils.doWithCommitAndFetchResult( - JobTemplateMetaMapper.class, - mapper -> mapper.updateJobTemplateMeta(newJobTemplatePO, oldJobTemplatePO)); + SessionUtils.doMultipleWithCommit( + () -> + OccWriteSupport.updateWithVersion( + () -> + SessionUtils.getWithoutCommit( + JobTemplateMetaMapper.class, + mapper -> + mapper.updateJobTemplateMeta(newJobTemplatePO, oldJobTemplatePO)), + () -> writeFailure(jobTemplateIdent, oldJobTemplatePO))); } catch (RuntimeException e) { - ExceptionUtils.checkSQLException( - e, Entity.EntityType.JOB_TEMPLATE, oldJobTemplateEntity.name()); + ExceptionUtils.checkSQLException(e, Entity.EntityType.JOB_TEMPLATE, jobTemplateIdent.name()); throw e; } - - if (result == null || result == 0) { - throw new NoSuchEntityException( - NoSuchEntityException.NO_SUCH_ENTITY_MESSAGE, - Entity.EntityType.JOB_TEMPLATE.name().toLowerCase(Locale.ROOT), - oldJobTemplateEntity.name()); - } else if (result > 1) { - throw new IOException( - String.format( - "Failed to update job template: %s, because more than one rows are updated: %d", - oldJobTemplateEntity.name(), result)); - } else { - return newJobTemplateEntity; - } + return newJobTemplateEntity; } private JobTemplatePO getJobTemplatePO(NameIdentifier jobTemplateIdent) { @@ -239,4 +221,62 @@ public List batchGetJobTemplateByIdentifier(List + OccWriteSupport.deleteWithVersion( + () -> + SessionUtils.getWithoutCommit( + JobTemplateMetaMapper.class, + mapper -> + mapper.softDeleteJobTemplateById( + observed.jobTemplateId(), observed.currentVersion())), + () -> writeFailure(ident, observed)), + () -> { + // The template CAS holds an exclusive lock, excluding new job inserts. A locking read + // also sees inserts committed while that CAS waited, even under REPEATABLE READ. + Long activeJob = + SessionUtils.getWithoutCommit( + JobMetaMapper.class, + mapper -> mapper.selectNonterminalJobForUpdate(observed.jobTemplateId())); + if (activeJob != null) { + throw new NonEmptyEntityException("Job template %s has active jobs", ident); + } + }, + () -> + SessionUtils.doWithoutCommit( + JobMetaMapper.class, + mapper -> mapper.softDeleteJobsByTemplateId(observed.jobTemplateId()))); + } + + /** Locks the observed template while a job is inserted in the same transaction. */ + void lockTemplateForJobWrite(String name, Long templateId, Long metalakeId) { + OccWriteSupport.lockParentForChildWrite( + name, + Entity.EntityType.JOB_TEMPLATE, + () -> + SessionUtils.getWithoutCommit( + JobTemplateMetaMapper.class, + mapper -> mapper.selectJobTemplateByIdForShare(templateId)), + null, + current -> + Objects.equals(current.jobTemplateName(), name) + && Objects.equals(current.metalakeId(), metalakeId)); + } + + private RuntimeException writeFailure(NameIdentifier ident, JobTemplatePO observed) { + return OccWriteSupport.writeFailure( + ident, + Entity.EntityType.JOB_TEMPLATE, + () -> + SessionUtils.getWithoutCommit( + JobTemplateMetaMapper.class, + mapper -> mapper.selectJobTemplateByIdForUpdate(observed.jobTemplateId())), + null, + current -> + Objects.equals(current.jobTemplateName(), observed.jobTemplateName()) + && Objects.equals(current.metalakeId(), observed.metalakeId())); + } } diff --git a/core/src/main/java/org/apache/gravitino/storage/relational/service/MetalakeMetaService.java b/core/src/main/java/org/apache/gravitino/storage/relational/service/MetalakeMetaService.java index 174cd97df18..4a2a0faa292 100644 --- a/core/src/main/java/org/apache/gravitino/storage/relational/service/MetalakeMetaService.java +++ b/core/src/main/java/org/apache/gravitino/storage/relational/service/MetalakeMetaService.java @@ -416,6 +416,19 @@ void deleteMetalakeWithVersion(NameIdentifier identifier, Long metalakeId, Long () -> metalakeWriteFailure(identifier, metalakeId, identifier.name())); } + /** Locks and validates a metalake while inserting a child in the current transaction. */ + void lockMetalakeForChildWrite(String name, Long metalakeId) { + OccWriteSupport.lockParentForChildWrite( + name, + Entity.EntityType.METALAKE, + () -> + SessionUtils.getWithoutCommit( + MetalakeMetaMapper.class, + mapper -> mapper.selectMetalakeMetaByIdForShare(metalakeId)), + null, + current -> Objects.equals(current.getMetalakeName(), name)); + } + private RuntimeException metalakeWriteFailure( NameIdentifier identifier, Long metalakeId, String observedName) { return OccWriteSupport.writeFailure( diff --git a/core/src/test/java/org/apache/gravitino/job/TestJobManager.java b/core/src/test/java/org/apache/gravitino/job/TestJobManager.java index ac88c2974ea..011281f3f0e 100644 --- a/core/src/test/java/org/apache/gravitino/job/TestJobManager.java +++ b/core/src/test/java/org/apache/gravitino/job/TestJobManager.java @@ -70,6 +70,8 @@ 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.LockManager; import org.apache.gravitino.meta.AuditInfo; @@ -172,6 +174,63 @@ public void tearDown() throws Exception { } } + @Test + public void testRegisterJobTemplateReportsConcurrentMetalakeDeletion() throws IOException { + JobTemplateEntity template = newShellJobTemplateEntity("shell_job", "A shell job template"); + NoSuchEntityException missing = new NoSuchEntityException("Metalake was deleted"); + doThrow(missing).when(entityStore).put(template, false); + + NoSuchMetalakeException failure = + Assertions.assertThrows( + NoSuchMetalakeException.class, + () -> jobManager.registerJobTemplate(metalake, template)); + Assertions.assertSame(missing, failure.getCause()); + } + + @Test + public void testRunJobReportsParentDisappearingDuringRegistration() throws Exception { + JobTemplateEntity template = newShellJobTemplateEntity("shell_job", "A shell job template"); + doReturn(template).when(jobManager).getJobTemplate(metalake, template.name()); + for (Entity.EntityType parent : + List.of(Entity.EntityType.METALAKE, Entity.EntityType.JOB_TEMPLATE)) { + Mockito.reset(entityStore, jobExecutor); + String executionId = "submitted_" + parent.name(); + when(jobExecutor.submitJob(any())).thenReturn(executionId); + NoSuchEntityException missing = new NoSuchEntityException("Parent was deleted: %s", parent); + doThrow(missing).when(entityStore).put(any(JobEntity.class), eq(false)); + + NoSuchJobTemplateException failure = + Assertions.assertThrows( + NoSuchJobTemplateException.class, + () -> jobManager.runJob(metalake, template.name(), Collections.emptyMap())); + Assertions.assertSame(missing, failure.getCause()); + verify(jobExecutor, times(1)).submitJob(any()); + verify(jobExecutor, never()).cancelJob(any()); + verify(entityStore, times(1)).put(any(JobEntity.class), eq(false)); + } + } + + @Test + public void testAlterJobTemplateDistinguishesMissingFromConflict() throws IOException { + NoSuchEntityException missing = new NoSuchEntityException("Template was deleted"); + doThrow(missing).when(entityStore).update(any(), any(), any(), any()); + NoSuchJobTemplateException failure = + Assertions.assertThrows( + NoSuchJobTemplateException.class, + () -> jobManager.alterJobTemplate(metalake, "shell_job")); + Assertions.assertEquals( + "Job template with name shell_job under metalake " + metalake + " does not exist", + failure.getMessage()); + + OptimisticLockException conflict = new OptimisticLockException("Template was modified"); + doThrow(conflict).when(entityStore).update(any(), any(), any(), any()); + Assertions.assertSame( + conflict, + Assertions.assertThrows( + OptimisticLockException.class, + () -> jobManager.alterJobTemplate(metalake, "shell_job"))); + } + @Test public void testListJobTemplates() throws IOException { mockedMetalake @@ -325,6 +384,86 @@ public void testGetJobTemplate() throws IOException { RuntimeException.class, () -> jobManager.getJobTemplate(metalake, "job")); } + /** A failed root CAS must not remove files belonging to the still-active template. */ + @Test + public void testDeleteJobTemplateConflictPreservesStaging() throws IOException { + JobEntity finishedJob = expiredJob(); + doReturn(Collections.singletonList(finishedJob)) + .when(jobManager) + .listJobs(metalake, Optional.of("shell_job")); + doThrow(new OptimisticLockException("template changed")) + .when(entityStore) + .delete( + NameIdentifierUtil.ofJobTemplate(metalake, "shell_job"), + Entity.EntityType.JOB_TEMPLATE); + File directory = + new File( + testStagingDir, + metalake + File.separator + "shell_job" + File.separator + finishedJob.name()); + Assertions.assertTrue(directory.mkdirs() || directory.isDirectory()); + File artifact = new File(directory, "artifact"); + Assertions.assertTrue(artifact.createNewFile()); + Assertions.assertThrows( + OptimisticLockException.class, () -> jobManager.deleteJobTemplate(metalake, "shell_job")); + Assertions.assertTrue(artifact.isFile()); + doReturn(true) + .when(entityStore) + .delete( + NameIdentifierUtil.ofJobTemplate(metalake, "shell_job"), + Entity.EntityType.JOB_TEMPLATE); + Assertions.assertTrue(jobManager.deleteJobTemplate(metalake, "shell_job")); + Assertions.assertFalse(directory.exists()); + } + + /** A successful delete must preserve files belonging to a same-name replacement. */ + @Test + public void testDeletePreservesReplacementStaging() throws IOException { + doReturn(Collections.emptyList()).when(jobManager).listJobs(metalake, Optional.of("shell_job")); + File replacementDir = + new File( + testStagingDir, metalake + File.separator + "shell_job" + File.separator + "job_999"); + File replacementArtifact = new File(replacementDir, "new-job-artifact"); + when(entityStore.delete( + NameIdentifierUtil.ofJobTemplate(metalake, "shell_job"), + Entity.EntityType.JOB_TEMPLATE)) + .thenAnswer( + invocation -> { + // The database delete has committed. Another server recreates the template and + // stages a new job before this server resumes its filesystem cleanup. + Assertions.assertTrue(replacementDir.mkdirs()); + Assertions.assertTrue(replacementArtifact.createNewFile()); + return true; + }); + + Assertions.assertTrue(jobManager.deleteJobTemplate(metalake, "shell_job")); + Assertions.assertTrue(replacementArtifact.isFile(), "Replacement job files must survive"); + } + + /** A job inserted after the initial check must prevent deletion without losing staging files. */ + @Test + public void testDeleteJobTemplateReportsConcurrentActiveJob() throws IOException { + JobEntity finishedJob = expiredJob(); + doReturn(Collections.singletonList(finishedJob)) + .when(jobManager) + .listJobs(metalake, Optional.of("shell_job")); + File directory = + new File( + testStagingDir, + metalake + File.separator + "shell_job" + File.separator + finishedJob.name()); + Assertions.assertTrue(directory.mkdirs()); + File artifact = new File(directory, "artifact"); + Assertions.assertTrue(artifact.createNewFile()); + doThrow(new NonEmptyEntityException("A job was inserted concurrently")) + .when(entityStore) + .delete( + NameIdentifierUtil.ofJobTemplate(metalake, "shell_job"), + Entity.EntityType.JOB_TEMPLATE); + + Assertions.assertThrows( + InUseException.class, () -> jobManager.deleteJobTemplate(metalake, "shell_job")); + Assertions.assertTrue(artifact.isFile()); + } + @Test public void testDeleteJobTemplate() throws IOException { mockedMetalake @@ -610,6 +749,19 @@ public void testRunJobSucceedsWhenStagingDirectoryAlreadyExists() throws Excepti } } + /** A metadata conflict must not replay the external cancellation operation. */ + @Test + public void testCancelJobDoesNotReplayExecutorOnOccConflict() throws IOException { + JobEntity job = newJobEntity("shell_job", JobHandle.Status.QUEUED); + when(jobManager.getJob(metalake, job.name())).thenReturn(job); + doNothing().when(jobExecutor).cancelJob(job.jobExecutionId()); + when(entityStore.update(any(), eq(JobEntity.class), eq(Entity.EntityType.JOB), any())) + .thenThrow(new OptimisticLockException("job changed")); + Assertions.assertThrows( + OptimisticLockException.class, () -> jobManager.cancelJob(metalake, job.name())); + verify(jobExecutor, times(1)).cancelJob(job.jobExecutionId()); + } + @Test public void testCancelJob() throws IOException { mockedMetalake @@ -1005,53 +1157,13 @@ public void testPullJobStatusStartedAtNotBackfilledOnDirectCancellation() throws @Test public void testPullJobStatusSkipsJobDeletedConcurrently() throws IOException { - JobEntity deletedJob = newJobEntity("shell_job", JobHandle.Status.QUEUED); - JobEntity survivingJob = newJobEntity("shell_job", JobHandle.Status.QUEUED); - - BaseMetalake mockMetalake = - BaseMetalake.builder() - .withName(metalake) - .withId(idGenerator.nextId()) - .withVersion(SchemaVersion.V_0_1) - .withAuditInfo(AuditInfo.EMPTY) - .build(); - when(entityStore.list(Namespace.empty(), BaseMetalake.class, Entity.EntityType.METALAKE)) - .thenReturn(ImmutableList.of(mockMetalake)); - mockedMetalake - .when(() -> MetalakeManager.listInUseMetalakes(entityStore)) - .thenReturn(ImmutableList.of(metalake)); - - when(jobManager.listJobs(metalake, Optional.empty())) - .thenReturn(ImmutableList.of(deletedJob, survivingJob)); - when(jobExecutor.getJobStatus(deletedJob.jobExecutionId())) - .thenReturn(JobHandle.Status.SUCCEEDED); - when(jobExecutor.getJobStatus(survivingJob.jobExecutionId())) - .thenReturn(JobHandle.Status.SUCCEEDED); - - // Simulate deletedJob having been removed from storage concurrently (e.g. by legacy-timeline - // cleanup) in the gap between the listJobs() snapshot above and the update call, while - // survivingJob's update succeeds normally. - NameIdentifier deletedJobIdent = NameIdentifierUtil.ofJob(metalake, deletedJob.name()); - NameIdentifier survivingJobIdent = NameIdentifierUtil.ofJob(metalake, survivingJob.name()); - when(entityStore.update( - eq(deletedJobIdent), eq(JobEntity.class), eq(Entity.EntityType.JOB), any())) - .thenThrow(new NoSuchEntityException("Job does not exist")); - when(entityStore.update( - eq(survivingJobIdent), eq(JobEntity.class), eq(Entity.EntityType.JOB), any())) - .thenAnswer( - invocation -> { - Function updater = invocation.getArgument(3); - return updater.apply(survivingJob); - }); - - // The disappearance of one job must not stop the rest of the batch from being processed, nor - // escape this method - scheduleAtFixedRate() would silently cancel all future runs otherwise. - Assertions.assertDoesNotThrow(() -> jobManager.pullAndUpdateJobStatus()); + assertStatusPollingContinues(new NoSuchEntityException("Job does not exist")); + } - verify(entityStore, times(1)) - .update(eq(deletedJobIdent), eq(JobEntity.class), eq(Entity.EntityType.JOB), any()); - verify(entityStore, times(1)) - .update(eq(survivingJobIdent), eq(JobEntity.class), eq(Entity.EntityType.JOB), any()); + /** Verifies OCC conflicts do not cancel future status polls. */ + @Test + public void testPullJobStatusContinuesAfterOccConflict() throws IOException { + assertStatusPollingContinues(new OptimisticLockException("Job changed concurrently")); } @Test @@ -1146,6 +1258,38 @@ public void testPullJobStatusDoesNotRegressConcurrentlyCancellingJob() throws IO Assertions.assertEquals(0L, result.finishedAt()); } + /** Conflicts preserve files without stopping this cleanup batch or its next scheduled run. */ + @Test + public void testCleanUpStagingDirsContinuesAfterOccConflict() throws IOException { + JobEntity conflicted = expiredJob(); + JobEntity other = expiredJob(); + mockedMetalake + .when(() -> MetalakeManager.listInUseMetalakes(entityStore)) + .thenReturn(ImmutableList.of(metalake)); + when(jobManager.listJobs(metalake, Optional.empty())) + .thenReturn(ImmutableList.of(conflicted, other), ImmutableList.of(conflicted)); + NameIdentifier conflictedIdent = NameIdentifierUtil.ofJob(metalake, conflicted.name()); + NameIdentifier otherIdent = NameIdentifierUtil.ofJob(metalake, other.name()); + when(entityStore.delete(conflictedIdent, Entity.EntityType.JOB)) + .thenThrow(new OptimisticLockException("job changed")) + .thenReturn(true); + when(entityStore.delete(otherIdent, Entity.EntityType.JOB)).thenReturn(true); + File conflictedDir = new File(testStagingDir, metalake + "/shell_job/" + conflicted.name()); + File otherDir = new File(testStagingDir, metalake + "/shell_job/" + other.name()); + Assertions.assertTrue(conflictedDir.mkdirs()); + Assertions.assertTrue(otherDir.mkdirs()); + File artifact = new File(conflictedDir, "artifact"); + Assertions.assertTrue(artifact.createNewFile()); + + Assertions.assertDoesNotThrow(() -> jobManager.cleanUpStagingDirs()); + Assertions.assertTrue(artifact.isFile()); + Assertions.assertFalse(otherDir.exists()); + Assertions.assertDoesNotThrow(() -> jobManager.cleanUpStagingDirs()); + Assertions.assertFalse(conflictedDir.exists()); + verify(entityStore, times(2)).delete(conflictedIdent, Entity.EntityType.JOB); + verify(entityStore, times(1)).delete(otherIdent, Entity.EntityType.JOB); + } + @Test public void testCleanUpStagingDirs() throws IOException, InterruptedException { JobEntity job = newJobEntity("shell_job", JobHandle.Status.STARTED); @@ -1400,6 +1544,54 @@ public void testUpdateSparkJobTemplateEntity() { oldJobTemplateEntity.nameIdentifier(), oldJobTemplateEntity, invalidChange)); } + private void assertStatusPollingContinues(RuntimeException failure) throws IOException { + JobEntity conflictedJob = newJobEntity("shell_job", JobHandle.Status.QUEUED); + JobEntity survivingJob = newJobEntity("shell_job", JobHandle.Status.QUEUED); + + BaseMetalake mockMetalake = + BaseMetalake.builder() + .withName(metalake) + .withId(idGenerator.nextId()) + .withVersion(SchemaVersion.V_0_1) + .withAuditInfo(AuditInfo.EMPTY) + .build(); + when(entityStore.list(Namespace.empty(), BaseMetalake.class, Entity.EntityType.METALAKE)) + .thenReturn(ImmutableList.of(mockMetalake)); + mockedMetalake + .when(() -> MetalakeManager.listInUseMetalakes(entityStore)) + .thenReturn(ImmutableList.of(metalake)); + + when(jobManager.listJobs(metalake, Optional.empty())) + .thenReturn(ImmutableList.of(conflictedJob, survivingJob)); + when(jobExecutor.getJobStatus(conflictedJob.jobExecutionId())) + .thenReturn(JobHandle.Status.SUCCEEDED); + when(jobExecutor.getJobStatus(survivingJob.jobExecutionId())) + .thenReturn(JobHandle.Status.SUCCEEDED); + + // A losing CAS must not stop this batch or future scheduled polls. + NameIdentifier conflictedJobIdent = NameIdentifierUtil.ofJob(metalake, conflictedJob.name()); + NameIdentifier survivingJobIdent = NameIdentifierUtil.ofJob(metalake, survivingJob.name()); + when(entityStore.update( + eq(conflictedJobIdent), eq(JobEntity.class), eq(Entity.EntityType.JOB), any())) + .thenThrow(failure); + when(entityStore.update( + eq(survivingJobIdent), eq(JobEntity.class), eq(Entity.EntityType.JOB), any())) + .thenAnswer( + invocation -> { + Function updater = invocation.getArgument(3); + return updater.apply(survivingJob); + }); + + // Both polls process the other job even when this job keeps conflicting. + Assertions.assertDoesNotThrow(() -> jobManager.pullAndUpdateJobStatus()); + Assertions.assertDoesNotThrow(() -> jobManager.pullAndUpdateJobStatus()); + + verify(entityStore, times(2)) + .update(eq(conflictedJobIdent), eq(JobEntity.class), eq(Entity.EntityType.JOB), any()); + verify(entityStore, times(2)) + .update(eq(survivingJobIdent), eq(JobEntity.class), eq(Entity.EntityType.JOB), any()); + } + private JobTemplateEntity newShellJobTemplateEntity(String name, String comment) { ShellJobTemplate shellJobTemplate = ShellJobTemplate.builder() @@ -1439,6 +1631,20 @@ private JobTemplateEntity newSparkJobTemplateEntity(String name, String comment) .build(); } + private JobEntity expiredJob() { + long id = idGenerator.nextId(); + return JobEntity.builder() + .withId(id) + .withJobExecutionId(Long.toString(id)) + .withNamespace(NamespaceUtil.ofJob(metalake)) + .withJobTemplateName("shell_job") + .withStartedAt(1L) + .withFinishedAt(2L) + .withStatus(JobHandle.Status.SUCCEEDED) + .withAuditInfo(AuditInfo.EMPTY) + .build(); + } + private JobEntity newJobEntity(String templateName, JobHandle.Status status) { Random rand = new Random(); return JobEntity.builder() diff --git a/core/src/test/java/org/apache/gravitino/storage/relational/service/TestJobTemplateMetaService.java b/core/src/test/java/org/apache/gravitino/storage/relational/service/TestJobTemplateMetaService.java index 1f40cadea24..7d3fec6f438 100644 --- a/core/src/test/java/org/apache/gravitino/storage/relational/service/TestJobTemplateMetaService.java +++ b/core/src/test/java/org/apache/gravitino/storage/relational/service/TestJobTemplateMetaService.java @@ -23,6 +23,7 @@ import java.util.List; import org.apache.gravitino.EntityAlreadyExistsException; import org.apache.gravitino.exceptions.NoSuchEntityException; +import org.apache.gravitino.exceptions.NonEmptyEntityException; import org.apache.gravitino.job.JobHandle; import org.apache.gravitino.meta.AuditInfo; import org.apache.gravitino.meta.BaseMetalake; @@ -196,6 +197,20 @@ public void testDeleteJobTemplateWithJobs() throws IOException { newJobEntity("job_template_with_jobs", JobHandle.Status.SUCCEEDED, METALAKE_NAME); backend.insert(jobEntity2, false); + Assertions.assertThrows( + NonEmptyEntityException.class, + () -> + jobTemplateMetaService.deleteJobTemplate( + NameIdentifierUtil.ofJobTemplate(METALAKE_NAME, "job_template_with_jobs"))); + Assertions.assertEquals( + 2, + JobMetaService.getInstance() + .listJobsByNamespace(NamespaceUtil.ofJob(METALAKE_NAME)) + .size()); + Assertions.assertTrue( + JobMetaService.getInstance() + .deleteJob(NameIdentifierUtil.ofJob(METALAKE_NAME, jobEntity1.name()))); + boolean deleted = jobTemplateMetaService.deleteJobTemplate( NameIdentifierUtil.ofJobTemplate(METALAKE_NAME, "job_template_with_jobs")); diff --git a/core/src/test/java/org/apache/gravitino/storage/relational/service/TestJobWriteOcc.java b/core/src/test/java/org/apache/gravitino/storage/relational/service/TestJobWriteOcc.java new file mode 100644 index 00000000000..ffe0652b727 --- /dev/null +++ b/core/src/test/java/org/apache/gravitino/storage/relational/service/TestJobWriteOcc.java @@ -0,0 +1,352 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.gravitino.storage.relational.service; + +import java.io.IOException; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; +import org.apache.gravitino.NameIdentifier; +import org.apache.gravitino.exceptions.NoSuchEntityException; +import org.apache.gravitino.exceptions.NonEmptyEntityException; +import org.apache.gravitino.exceptions.OptimisticLockException; +import org.apache.gravitino.job.JobHandle; +import org.apache.gravitino.meta.JobEntity; +import org.apache.gravitino.meta.JobTemplateEntity; +import org.apache.gravitino.storage.RandomIdGenerator; +import org.apache.gravitino.storage.relational.TestJDBCBackend; +import org.apache.gravitino.storage.relational.mapper.JobMetaMapper; +import org.apache.gravitino.storage.relational.mapper.JobTemplateMetaMapper; +import org.apache.gravitino.storage.relational.mapper.MetalakeMetaMapper; +import org.apache.gravitino.storage.relational.po.JobPO; +import org.apache.gravitino.storage.relational.po.JobTemplatePO; +import org.apache.gravitino.storage.relational.utils.SessionUtils; +import org.apache.gravitino.utils.NameIdentifierUtil; +import org.apache.gravitino.utils.NamespaceUtil; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.TestTemplate; +import org.junit.jupiter.api.function.Executable; + +/** Exercises job and template OCC against each configured relational backend. */ +public class TestJobWriteOcc extends TestJDBCBackend { + private static final String METALAKE = "job_occ"; + private final JobMetaService jobs = JobMetaService.getInstance(); + private final JobTemplateMetaService templates = JobTemplateMetaService.getInstance(); + private JobTemplateEntity template; + private JobEntity job; + private long metalakeId; + + /** A version-only change defeats stale updates and deletes without hiding the entity. */ + @TestTemplate + public void testJobConflictsAndIdempotentDelete() throws IOException { + initialize(); + JobPO observed = jobPO(); + Assertions.assertThrows( + OptimisticLockException.class, + () -> + jobs.updateJob( + jobIdent(), + old -> { + Assertions.assertDoesNotThrow( + () -> jobs.updateJob(jobIdent(), current -> current)); + return old; + })); + Assertions.assertEquals(observed.currentVersion() + 1, jobPO().currentVersion()); + Assertions.assertThrows( + OptimisticLockException.class, () -> jobs.deleteJobWithVersion(jobIdent(), observed)); + Assertions.assertTrue(jobs.deleteJob(jobIdent())); + Assertions.assertFalse(jobs.deleteJob(jobIdent())); + Assertions.assertThrows( + NoSuchEntityException.class, () -> jobs.updateJob(jobIdent(), current -> current)); + } + + /** A delete winning after the update read prevents resurrection. */ + @TestTemplate + public void testDeleteWinsOverJobUpdate() throws IOException { + initialize(); + Assertions.assertThrows( + NoSuchEntityException.class, + () -> + jobs.updateJob( + jobIdent(), + old -> { + Assertions.assertTrue(jobs.deleteJob(jobIdent())); + return old; + })); + Assertions.assertFalse(jobs.deleteJob(jobIdent())); + } + + /** Failed template CAS leaves child jobs untouched; a current delete removes both. */ + @TestTemplate + public void testTemplateConflictsBeforeCascade() throws IOException { + initialize(); + JobTemplatePO observed = templatePO(); + Assertions.assertThrows( + OptimisticLockException.class, + () -> + templates.updateJobTemplate( + templateIdent(), + old -> { + Assertions.assertDoesNotThrow( + () -> templates.updateJobTemplate(templateIdent(), current -> current)); + return old; + })); + Assertions.assertEquals(observed.currentVersion() + 1, templatePO().currentVersion()); + Assertions.assertThrows( + OptimisticLockException.class, + () -> templates.deleteJobTemplateWithVersion(templateIdent(), observed)); + Assertions.assertEquals(job.id(), jobs.getJobByIdentifier(jobIdent()).id()); + Assertions.assertTrue(templates.deleteJobTemplate(templateIdent())); + Assertions.assertFalse(templates.deleteJobTemplate(templateIdent())); + Assertions.assertTrue(jobs.listJobsByNamespace(NamespaceUtil.ofJob(METALAKE)).isEmpty()); + } + + /** Rollback restores the template and its already-deleted jobs. */ + @TestTemplate + public void testCascadeRollback() throws IOException { + initialize(); + JobTemplatePO observed = templatePO(); + Assertions.assertThrows( + IllegalStateException.class, + () -> + SessionUtils.doMultipleWithCommit( + () -> templates.deleteJobTemplateWithVersion(templateIdent(), observed), + () -> { + throw new IllegalStateException("injected failure after cascade"); + })); + Assertions.assertEquals(observed.currentVersion(), templatePO().currentVersion()); + Assertions.assertEquals(job.id(), jobs.getJobByIdentifier(jobIdent()).id()); + } + + /** A stale snapshot cannot delete a same-name replacement or its jobs. */ + @TestTemplate + public void testSameNameRecreation() throws IOException { + initialize(); + JobTemplatePO observed = templatePO(); + templates.deleteJobTemplate(templateIdent()); + template = TestJobTemplateMetaService.newShellJobTemplateEntity("template", "new", METALAKE); + templates.insertJobTemplate(template, false); + job = TestJobTemplateMetaService.newJobEntity("template", JobHandle.Status.QUEUED, METALAKE); + jobs.insertJob(job, false); + Assertions.assertThrows( + NoSuchEntityException.class, + () -> templates.deleteJobTemplateWithVersion(templateIdent(), observed)); + Assertions.assertEquals(job.id(), jobs.getJobByIdentifier(jobIdent()).id()); + } + + /** Job insertion waits for an in-flight template delete and then fails without inserting. */ + @TestTemplate + public void testJobInsertFencedByTemplateDelete() throws Exception { + initialize(); + JobTemplatePO observed = templatePO(); + JobEntity candidate = + TestJobTemplateMetaService.newJobEntity("template", JobHandle.Status.QUEUED, METALAKE); + Throwable failure = + whileWriteUncommitted( + () -> templates.deleteJobTemplateWithVersion(templateIdent(), observed), + () -> jobs.insertJob(candidate, false)); + Assertions.assertInstanceOf(NoSuchEntityException.class, failure); + Assertions.assertNull( + SessionUtils.getWithoutCommit( + JobMetaMapper.class, + mapper -> mapper.selectJobRunIdForUpdate(candidate.id(), metalakeId))); + } + + /** Metalake fencing rejects both kinds of insert after parent deletion commits. */ + @TestTemplate + public void testInsertsFencedByMetalakeDelete() throws Exception { + initialize(); + JobTemplateEntity candidate = + TestJobTemplateMetaService.newShellJobTemplateEntity("other", "new", METALAKE); + Throwable failure = + whileWriteUncommitted( + () -> + SessionUtils.doWithoutCommit( + MetalakeMetaMapper.class, + mapper -> mapper.softDeleteMetalakeMetaByMetalakeId(metalakeId, 1L)), + () -> templates.insertJobTemplate(candidate, false)); + Assertions.assertInstanceOf(NoSuchEntityException.class, failure); + Assertions.assertThrows(NoSuchEntityException.class, () -> jobs.insertJob(job, false)); + } + + /** Missing templates are reported instead of silently dropping job insertion failures. */ + @TestTemplate + public void testMissingTemplateInsertFails() throws IOException { + initialize(); + JobEntity candidate = + TestJobTemplateMetaService.newJobEntity("missing", JobHandle.Status.QUEUED, METALAKE); + Assertions.assertThrows(NoSuchEntityException.class, () -> jobs.insertJob(candidate, false)); + } + + /** A parent delete waits for an in-flight insertion and preserves the committed active job. */ + @TestTemplate + public void testTemplateDeleteWaitsForJobInsert() throws Exception { + initialize(); + JobEntity candidate = + TestJobTemplateMetaService.newJobEntity("template", JobHandle.Status.QUEUED, METALAKE); + Throwable failure = + whileWriteUncommitted( + () -> Assertions.assertDoesNotThrow(() -> jobs.insertJob(candidate, false)), + () -> templates.deleteJobTemplate(templateIdent())); + Assertions.assertInstanceOf(NonEmptyEntityException.class, failure); + Assertions.assertEquals(template.id(), templatePO().jobTemplateId()); + Assertions.assertEquals(1L, templatePO().currentVersion()); + Assertions.assertEquals( + candidate.id(), + jobs.getJobByIdentifier(NameIdentifierUtil.ofJob(METALAKE, candidate.name())).id()); + } + + /** An identifier under another metalake cannot delete a job solely by its numeric run ID. */ + @TestTemplate + public void testDeleteChecksMetalakeIdentity() throws IOException { + initialize(); + Assertions.assertFalse(jobs.deleteJob(NameIdentifierUtil.ofJob("other", job.name()))); + Assertions.assertEquals(job.id(), jobs.getJobByIdentifier(jobIdent()).id()); + } + + /** Renaming invalidates the old name while cascade cleanup continues to use stable IDs. */ + @TestTemplate + public void testRenameAndCascadeIdentity() throws IOException { + initialize(); + NameIdentifier oldIdent = templateIdent(); + JobTemplatePO observed = templatePO(); + template = + templates.updateJobTemplate( + oldIdent, + old -> + JobTemplateEntity.builder() + .withId(old.id()) + .withName("renamed") + .withNamespace(old.namespace()) + .withComment(old.comment()) + .withTemplateContent(old.templateContent()) + .withAuditInfo(old.auditInfo()) + .build()); + Assertions.assertThrows( + NoSuchEntityException.class, + () -> templates.deleteJobTemplateWithVersion(oldIdent, observed)); + Assertions.assertEquals("renamed", jobs.getJobByIdentifier(jobIdent()).jobTemplateName()); + Assertions.assertTrue(templates.deleteJobTemplate(templateIdent())); + Assertions.assertNull( + SessionUtils.getWithoutCommit( + JobMetaMapper.class, mapper -> mapper.selectJobRunIdForUpdate(job.id(), metalakeId))); + } + + /** All nonterminal states reject deletion and roll back the root CAS. */ + @TestTemplate + public void testNonterminalJobsPreventTemplateDeletion() throws IOException { + initialize(); + for (JobHandle.Status status : + new JobHandle.Status[] { + JobHandle.Status.QUEUED, JobHandle.Status.STARTED, JobHandle.Status.CANCELLING + }) { + JobEntity active = TestJobTemplateMetaService.newJobEntity("template", status, METALAKE); + jobs.insertJob(active, false); + Assertions.assertThrows( + NonEmptyEntityException.class, () -> templates.deleteJobTemplate(templateIdent())); + Assertions.assertEquals(1L, templatePO().currentVersion()); + Assertions.assertEquals(job.id(), jobs.getJobByIdentifier(jobIdent()).id()); + Assertions.assertEquals( + active.id(), + jobs.getJobByIdentifier(NameIdentifierUtil.ofJob(METALAKE, active.name())).id()); + Assertions.assertTrue(jobs.deleteJob(NameIdentifierUtil.ofJob(METALAKE, active.name()))); + } + for (JobHandle.Status status : + new JobHandle.Status[] {JobHandle.Status.CANCELLED, JobHandle.Status.FAILED}) { + jobs.insertJob(TestJobTemplateMetaService.newJobEntity("template", status, METALAKE), false); + } + Assertions.assertTrue(templates.deleteJobTemplate(templateIdent())); + Assertions.assertTrue(jobs.listJobsByNamespace(NamespaceUtil.ofJob(METALAKE)).isEmpty()); + } + + private void initialize() throws IOException { + metalakeId = RandomIdGenerator.INSTANCE.nextId(); + backend.insert(createBaseMakeLake(metalakeId, METALAKE, AUDIT_INFO), false); + template = + TestJobTemplateMetaService.newShellJobTemplateEntity("template", "original", METALAKE); + templates.insertJobTemplate(template, false); + job = TestJobTemplateMetaService.newJobEntity("template", JobHandle.Status.SUCCEEDED, METALAKE); + jobs.insertJob(job, false); + } + + private NameIdentifier jobIdent() { + return NameIdentifierUtil.ofJob(METALAKE, job.name()); + } + + private NameIdentifier templateIdent() { + return NameIdentifierUtil.ofJobTemplate(METALAKE, template.name()); + } + + private JobPO jobPO() { + return SessionUtils.getWithoutCommit( + JobMetaMapper.class, mapper -> mapper.selectJobPOByMetalakeAndRunId(METALAKE, job.id())); + } + + private JobTemplatePO templatePO() { + return SessionUtils.getWithoutCommit( + JobTemplateMetaMapper.class, mapper -> mapper.selectJobTemplateById(template.id())); + } + + private Throwable whileWriteUncommitted(Runnable write, Executable victim) throws Exception { + CountDownLatch locked = new CountDownLatch(1); + CountDownLatch commit = new CountDownLatch(1); + CountDownLatch started = new CountDownLatch(1); + ExecutorService executor = Executors.newFixedThreadPool(2); + try { + Future writer = + executor.submit( + () -> + SessionUtils.doMultipleWithCommit( + write, + () -> { + locked.countDown(); + try { + Assertions.assertTrue(commit.await(30, TimeUnit.SECONDS)); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new RuntimeException(e); + } + })); + Assertions.assertTrue(locked.await(30, TimeUnit.SECONDS)); + Future reader = + executor.submit( + () -> { + started.countDown(); + try { + victim.execute(); + return null; + } catch (Throwable t) { + return t; + } + }); + Assertions.assertTrue(started.await(30, TimeUnit.SECONDS)); + Assertions.assertThrows(TimeoutException.class, () -> reader.get(500, TimeUnit.MILLISECONDS)); + commit.countDown(); + writer.get(30, TimeUnit.SECONDS); + return reader.get(30, TimeUnit.SECONDS); + } finally { + commit.countDown(); + executor.shutdownNow(); + Assertions.assertTrue(executor.awaitTermination(30, TimeUnit.SECONDS)); + } + } +} diff --git a/server/src/test/java/org/apache/gravitino/server/web/rest/TestExceptionHandlers.java b/server/src/test/java/org/apache/gravitino/server/web/rest/TestExceptionHandlers.java index 07d56d6ea82..ff03f3462b4 100644 --- a/server/src/test/java/org/apache/gravitino/server/web/rest/TestExceptionHandlers.java +++ b/server/src/test/java/org/apache/gravitino/server/web/rest/TestExceptionHandlers.java @@ -22,6 +22,9 @@ import javax.ws.rs.core.Response; import org.apache.gravitino.dto.responses.ErrorConstants; import org.apache.gravitino.dto.responses.ErrorResponse; +import org.apache.gravitino.exceptions.InUseException; +import org.apache.gravitino.exceptions.NoSuchJobTemplateException; +import org.apache.gravitino.exceptions.NoSuchMetalakeException; import org.apache.gravitino.exceptions.OptimisticLockException; import org.apache.gravitino.exceptions.UnmodifiableStatisticException; import org.junit.jupiter.api.Assertions; @@ -29,6 +32,46 @@ public class TestExceptionHandlers { + @Test + void testConcurrentActiveJobReturnsConflict() { + try (Response response = + ExceptionHandlers.handleJobTemplateException( + OperationType.DELETE, + "template", + "metalake", + new InUseException("Template has an active job"))) { + Assertions.assertEquals(Response.Status.CONFLICT.getStatusCode(), response.getStatus()); + Assertions.assertEquals( + InUseException.class.getSimpleName(), ((ErrorResponse) response.getEntity()).getType()); + } + } + + @Test + public void testMissingJobParentsReturnNotFound() { + Response runResponse = + ExceptionHandlers.handleJobException( + OperationType.RUN, + "template", + "metalake", + new NoSuchJobTemplateException("Template disappeared")); + Assertions.assertEquals(Response.Status.NOT_FOUND.getStatusCode(), runResponse.getStatus()); + Assertions.assertEquals( + NoSuchJobTemplateException.class.getSimpleName(), + ((ErrorResponse) runResponse.getEntity()).getType()); + + Response registerResponse = + ExceptionHandlers.handleJobTemplateException( + OperationType.REGISTER, + "template", + "metalake", + new NoSuchMetalakeException("Metalake disappeared")); + Assertions.assertEquals( + Response.Status.NOT_FOUND.getStatusCode(), registerResponse.getStatus()); + Assertions.assertEquals( + NoSuchMetalakeException.class.getSimpleName(), + ((ErrorResponse) registerResponse.getEntity()).getType()); + } + @Test public void testGetErrorMsg() { Exception e1 = new Exception("test1");