This is an automated email from the ASF dual-hosted git repository.
jerryshao pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/gravitino.git
The following commit(s) were added to refs/heads/main by this push:
new faae94aba3 [#12992]feat(core): complete OCC for jobs and job templates
(#12994)
faae94aba3 is described below
commit faae94aba346a76ae3a6eb990c66612b1daed8c4
Author: Qi Yu <[email protected]>
AuthorDate: Fri Sep 11 15:24:52 2026 +0800
[#12992]feat(core): complete OCC for jobs and job templates (#12994)
### What changes were proposed in this pull request?
Complete version-CAS updates and soft-deletes for Job and Job Template
using the shared OCC helpers. Preserve missing-entity and
idempotent-delete behavior while reporting stale writes as
optimistic-lock conflicts.
Fence job/template insertion with parent row locks. Delete the template
root by expected version, check for nonterminal jobs using a locking
read, and cascade by stable template ID in one transaction. If a
concurrent insertion commits first, reject deletion and roll back the
root change.
Keep background polling and staging cleanup alive after conflicts.
Translate disappearing parents into not-found exceptions. After template
deletion succeeds, clean only the observed job directories so a
same-name replacement's files survive. Retain the template parent
directory rather than recursively deleting it.
Reuse metalake fencing and terminal-state checks, and fetch only
identity columns for template locking reads.
### Why are the changes needed?
Unversioned deletes and unfenced inserts can leave orphan metadata or
remove child jobs after a failed template deletion. Checking for active
jobs outside the deletion transaction misses concurrent inserts.
Removing the entire template-name directory after committing can delete
files created by a same-name replacement.
Fix: #12992
Fix: #12993
### Does this PR introduce _any_ user-facing change?
Stale writes use the existing OCC conflict contract. Concurrent active
jobs prevent template deletion with `InUseException` (409). Failed
deletes preserve staging files, and successful deletes do not
recursively remove a replacement template's directory.
No public API signatures, configuration keys, schema, job identifiers,
runtime-template JSON, or blind create/import overwrite semantics
change. External submission and cancellation are not retried; submission
compensation remains tracked in #10271.
### How was this patch tested?
- 119 Core tests passed with no failures or skips: `TestJobManager`,
`TestJobMetaService`, `TestJobTemplateMetaService`, and
`TestJobWriteOcc`. All relational tests ran against H2, MySQL, and
PostgreSQL (`-PskipITs -PskipDockerTests=false`).
- All 6 `TestExceptionHandlers` tests passed, including the active-job
409 response.
- `./gradlew spotlessApply` and `git diff --check` passed.
- Core/Server checks passed with the already-run Core tests excluded:
`./gradlew :core:spotlessApply :server:spotlessApply :core:check
:server:test --tests
org.apache.gravitino.server.web.rest.TestExceptionHandlers :server:check
-PskipITs -PskipDockerTests=true -x :core:test`.
Coverage includes stale writes/deletes, same-name recreation, template
rename, parent fencing, cascade rollback, polling/cleanup continuity,
cancellation without replay, missing-parent errors, all nonterminal
states, and concurrent insertion preventing template deletion. A
controlled-interleaving regression reproduced replacement staging-file
deletion before the fix.
The full repository suite and external job-executor deployment tests
were not run.
---------
Signed-off-by: yuqi <[email protected]>
---
.../java/org/apache/gravitino/job/JobManager.java | 104 ++++--
.../storage/relational/mapper/JobMetaMapper.java | 52 ++-
.../mapper/JobMetaSQLProviderFactory.java | 54 +++-
.../relational/mapper/JobTemplateMetaMapper.java | 43 ++-
.../mapper/JobTemplateMetaSQLProviderFactory.java | 44 ++-
.../provider/base/JobMetaBaseSQLProvider.java | 84 +++--
.../base/JobTemplateMetaBaseSQLProvider.java | 65 +++-
.../postgresql/JobMetaPostgreSQLProvider.java | 47 ++-
.../JobTemplateMetaPostgreSQLProvider.java | 31 +-
.../storage/relational/service/JobMetaService.java | 98 ++++--
.../relational/service/JobTemplateMetaService.java | 138 +++++---
.../relational/service/MetalakeMetaService.java | 13 +
.../org/apache/gravitino/job/TestJobManager.java | 298 ++++++++++++++---
.../service/TestJobTemplateMetaService.java | 15 +
.../relational/service/TestJobWriteOcc.java | 352 +++++++++++++++++++++
.../server/web/rest/TestExceptionHandlers.java | 43 +++
16 files changed, 1198 insertions(+), 283 deletions(-)
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 6db68afb14..e8365cda86 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.JobTemplateAlreadyExistsException;
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 class JobManager implements JobOperationDispatcher {
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 class JobManager implements JobOperationDispatcher
{
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 class JobManager implements JobOperationDispatcher {
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 class JobManager implements JobOperationDispatcher {
try {
entityStore.put(jobEntity, false /* overwrite */);
+ } catch (NoSuchEntityException e) {
+ LOG.error(
+ "Job {} was submitted as execution {} but could not be registered
because its template "
+ + "{} or metalake {} no longer exists",
+ jobEntity.name(),
+ jobExecutionId,
+ jobTemplateName,
+ metalake,
+ e);
+ throw new NoSuchJobTemplateException(
+ e,
+ "Job template with name %s under metalake %s does not exist",
+ jobTemplateName,
+ metalake);
} catch (IOException e) {
throw new RuntimeException("Failed to register the job entity " +
jobEntity, e);
}
@@ -669,6 +692,14 @@ public class JobManager implements JobOperationDispatcher {
e);
}
});
+ } catch (OptimisticLockException e) {
+ // A later poll re-reads both executor state and metadata.
Never stop the scheduled
+ // task or replay external submission/cancellation because a
metadata CAS lost.
+ LOG.info(
+ "Job {} under metalake {} changed concurrently; deferring
status update",
+ job.name(),
+ metalake);
+ return;
} catch (NoSuchEntityException e) {
// The job could have been deleted concurrently (e.g. by
legacy-timeline cleanup)
// in the gap between the listJobs() snapshot above and this
update. Skip it rather
@@ -767,11 +798,7 @@ public class JobManager implements JobOperationDispatcher {
for (String metalake : metalakes) {
List<JobEntity> finishedJobs =
listJobs(metalake, Optional.empty()).stream()
- .filter(
- job ->
- job.status() == JobHandle.Status.CANCELLED
- || job.status() == JobHandle.Status.SUCCEEDED
- || job.status() == JobHandle.Status.FAILED)
+ .filter(job -> isFinishedStatus(job.status()))
.filter(
job ->
job.finishedAt() > 0
@@ -793,6 +820,13 @@ public class JobManager implements JobOperationDispatcher {
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 ae667fcad1..c11d1a92e3 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 @@ public interface JobMetaMapper {
@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 @@ public interface JobMetaMapper {
Integer deleteJobMetasByLegacyTimeline(
@Param("legacyTimeline") Long legacyTimeline, @Param("limit") int limit);
- @UpdateProvider(type = JobMetaSQLProviderFactory.class, method =
"softDeleteJobMetaByRunId")
- Integer softDeleteJobMetaByRunId(@Param("jobRunId") Long jobRunId);
-
@SelectProvider(type = JobMetaSQLProviderFactory.class, method =
"batchSelectJobByRunIds")
List<JobPO> batchSelectJobByRunIds(
@Param("metalakeName") String metalakeName, @Param("jobRunIds")
List<Long> jobRunIds);
+ /**
+ * Locks the active row for OCC identity validation.
+ *
+ * @param jobRunId the stable job run ID
+ * @param metalakeId the owning metalake ID
+ * @return the active row identity, or null if missing
+ */
+ @Nullable
+ @SelectProvider(type = JobMetaSQLProviderFactory.class, method =
"selectJobRunIdForUpdate")
+ Long selectJobRunIdForUpdate(
+ @Param("jobRunId") Long jobRunId, @Param("metalakeId") Long metalakeId);
+
+ /**
+ * Deletes active metadata using a stable identity and expected version.
+ *
+ * @param jobRunId the stable job run ID
+ * @param currentVersion the expected OCC version
+ * @return the affected row count
+ */
+ @UpdateProvider(
+ type = JobMetaSQLProviderFactory.class,
+ method = "softDeleteJobByRunIdWithVersion")
+ int softDeleteJobByRunIdWithVersion(
+ @Param("jobRunId") Long jobRunId, @Param("currentVersion") Long
currentVersion);
+
+ /**
+ * Deletes active metadata using a stable identity.
+ *
+ * @param jobTemplateId the stable template ID
+ * @return the affected row count
+ */
+ @UpdateProvider(type = JobMetaSQLProviderFactory.class, method =
"softDeleteJobsByTemplateId")
+ int softDeleteJobsByTemplateId(@Param("jobTemplateId") Long jobTemplateId);
+ /**
+ * Locks a nonterminal job belonging to the template using a current
database read.
+ *
+ * @param jobTemplateId the stable template ID
+ * @return a nonterminal job ID, or null if there are none
+ */
+ @Nullable
+ @SelectProvider(type = JobMetaSQLProviderFactory.class, method =
"selectNonterminalJobForUpdate")
+ Long selectNonterminalJobForUpdate(@Param("jobTemplateId") Long
jobTemplateId);
}
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 20d6c414e1..f75ce6e18f 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 class JobMetaSQLProviderFactory {
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 class JobMetaSQLProviderFactory {
return getProvider().softDeleteJobMetasByLegacyTimeline(legacyTimeline);
}
- public static String softDeleteJobMetaByRunId(@Param("jobRunId") Long
jobRunId) {
- return getProvider().softDeleteJobMetaByRunId(jobRunId);
- }
-
public static String batchSelectJobByRunIds(
@Param("metalakeName") String metalakeName, @Param("jobRunIds")
List<Long> jobRunIds) {
return getProvider().batchSelectJobByRunIds(metalakeName, jobRunIds);
}
+
+ /**
+ * Locks the active row for OCC identity validation.
+ *
+ * @param jobRunId the stable job run ID
+ * @param metalakeId the owning metalake ID
+ * @return the SQL statement
+ */
+ public static String selectJobRunIdForUpdate(
+ @Param("jobRunId") Long jobRunId, @Param("metalakeId") Long metalakeId) {
+ return getProvider().selectJobRunIdForUpdate(jobRunId, metalakeId);
+ }
+
+ /**
+ * Deletes active metadata using a stable identity and expected version.
+ *
+ * @param jobRunId the stable job run ID
+ * @param currentVersion the expected OCC version
+ * @return the SQL statement
+ */
+ public static String softDeleteJobByRunIdWithVersion(
+ @Param("jobRunId") Long jobRunId, @Param("currentVersion") Long
currentVersion) {
+ return getProvider().softDeleteJobByRunIdWithVersion(jobRunId,
currentVersion);
+ }
+
+ /**
+ * Deletes active metadata using a stable identity.
+ *
+ * @param jobTemplateId the stable template ID
+ * @return the SQL statement
+ */
+ public static String softDeleteJobsByTemplateId(@Param("jobTemplateId") Long
jobTemplateId) {
+ return getProvider().softDeleteJobsByTemplateId(jobTemplateId);
+ }
+
+ /**
+ * Builds a locking lookup for a nonterminal job belonging to a template.
+ *
+ * @param jobTemplateId the stable template ID
+ * @return the SQL statement
+ */
+ public static String selectNonterminalJobForUpdate(@Param("jobTemplateId")
Long jobTemplateId) {
+ return getProvider().selectNonterminalJobForUpdate(jobTemplateId);
+ }
}
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 09f5d52f62..d17521b2c8 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 @@ public interface JobTemplateMetaMapper {
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 @@ public interface JobTemplateMetaMapper {
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);
}
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 04383da401..8e59c1d0b4 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 class JobTemplateMetaSQLProviderFactory {
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 class JobTemplateMetaSQLProviderFactory {
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 class JobTemplateMetaSQLProviderFactory {
@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);
+ }
}
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 344f4007fd..89ffd79aed 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 class JobMetaBaseSQLProvider {
+ " 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 class JobMetaBaseSQLProvider {
+ " 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 class JobMetaBaseSQLProvider {
+ " AND jrm.deleted_at = 0 AND jtm.deleted_at = 0 AND mm.deleted_at =
0"
+ "</script>";
}
+
+ /**
+ * 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 747e98e6e0..002c57b647 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 class JobTemplateMetaBaseSQLProvider {
+ " 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 class JobTemplateMetaBaseSQLProvider {
+ " 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 class JobTemplateMetaBaseSQLProvider {
+ " AND jtm.deleted_at = 0 AND mm.deleted_at = 0"
+ "</script>";
}
+
+ /**
+ * 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 f2169ca4e9..1e73489f09 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 @@ package
org.apache.gravitino.storage.relational.mapper.provider.postgresql;
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 class JobMetaPostgreSQLProvider extends
JobMetaBaseSQLProvider {
+ " 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 class JobMetaPostgreSQLProvider extends
JobMetaBaseSQLProvider {
+ " 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 class JobMetaPostgreSQLProvider extends
JobMetaBaseSQLProvider {
+ 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 e7426ff45b..50db19ca31 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 @@ import org.apache.ibatis.annotations.Param;
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 class JobTemplateMetaPostgreSQLProvider extends
JobTemplateMetaBaseSQLPro
+ 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 a67b8a21f3..a07a13ce0f 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 class JobMetaService {
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 class JobMetaService {
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 class JobMetaService {
.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 15685af492..9f5dffbfa8 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.io.IOException;
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.HasIdentifier;
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 class JobTemplateMetaService {
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 class JobTemplateMetaService {
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 class JobTemplateMetaService {
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 class JobTemplateMetaService {
.collect(Collectors.toList());
});
}
+
+ /** Deletes the observed template before its jobs, rolling back all changes
on any failure. */
+ void deleteJobTemplateWithVersion(NameIdentifier ident, JobTemplatePO
observed) {
+ SessionUtils.doMultipleWithCommit(
+ () ->
+ 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 174cd97df1..4a2a0faa29 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 @@ public class MetalakeMetaService {
() -> 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 ac88c2974e..011281f3f0 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.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.LockManager;
import org.apache.gravitino.meta.AuditInfo;
@@ -172,6 +174,63 @@ public class TestJobManager {
}
}
+ @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 class TestJobManager {
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 class TestJobManager {
}
}
+ /** 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 class TestJobManager {
@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<JobEntity, JobEntity> 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 class TestJobManager {
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 class TestJobManager {
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<JobEntity, JobEntity> 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 @@ public class TestJobManager {
.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 1f40cadea2..7d3fec6f43 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.time.Instant;
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 class TestJobTemplateMetaService extends
TestJDBCBackend {
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 0000000000..ffe0652b72
--- /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.<JobEntity>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.<JobEntity>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.<JobTemplateEntity>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.<JobTemplateEntity>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<Throwable> 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 07d56d6ea8..ff03f3462b 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 java.util.List;
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 @@ import org.junit.jupiter.api.Test;
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");
