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 12ddb87894 [#12669] improvement(core): Use entityStore.update() for
job status transitions (#12675)
12ddb87894 is described below
commit 12ddb878948ddacea29f66754fb9d81ffa607ce9
Author: Jerry Shao <[email protected]>
AuthorDate: Fri Aug 28 19:02:13 2026 +0800
[#12669] improvement(core): Use entityStore.update() for job status
transitions (#12675)
### What changes were proposed in this pull request?
Replaces the `entityStore.put(entity, overwrite=true)` + manual
re-fetch-under-lock pattern in
`JobManager.cancelJob`/`pullAndUpdateJobStatus` with
`entityStore.update(...)`, matching the pattern already used for
`JOB_TEMPLATE` and other entities:
- Adds `JobMetaService.updateJob` (plus the `JDBCBackend` dispatch case,
the `updateJobMeta` mapper/SQL provider method, and `JobPO.updateJobPO`)
needed to route `JOB` entity writes through `entityStore.update()`.
- `updateJobMeta`'s `WHERE` clause now does a proper
`current_version`/`last_version` compare-and-swap, matching the same
pattern used by `View`/`Table`/`JobTemplate`.
- `JobManager.cancelJob` now translates a `NoSuchEntityException` from a
concurrently deleted job into `NoSuchJobException` (matching `getJob`'s
existing behavior) instead of leaking an undeclared exception.
- `JobManager.pullAndUpdateJobStatus` now catches
`NoSuchEntityException` per-job and skips it (logging a warning) instead
of letting the exception escape the scheduled task — previously this
would have caused `ScheduledExecutorService` to silently and permanently
cancel all future status-poll runs.
- Also includes an incidental, unrelated commit reformatting two long
`assert_properties_equal` calls in the Python integration tests (no
behavior change).
### Why are the changes needed?
The job entity was the only entity type still using a raw
`put(overwrite)` for updates, requiring every call site to manually
re-fetch the latest state under a lock before overwriting — a pattern
that's easy to get wrong (see the removed comment in `cancelJob`
describing a near-miss with a stale snapshot) and inconsistent with
every other entity in `JDBCBackend`.
Fix: #12669
### Does this PR introduce _any_ user-facing change?
No user-facing/API change. Internal storage-layer behavior only.
### How was this patch tested?
Added/updated unit tests:
- `TestJobMetaService`: `testUpdateJob`,
`testUpdateNonExistentJobThrowsNoSuchEntityException`,
`testUpdateJobWithMismatchedIdThrowsIllegalArgumentException`,
`testUpdateJobWithMalformedIdentifierThrowsNoSuchEntityException`.
- `TestJobPO`: `testUpdateJobPO`.
- `TestJobManager`: updated existing
`cancelJob`/`pullAndUpdateJobStatus` tests to mock
`entityStore.update()` instead of `put()`; added
`testCancelJobThrowsNoSuchJobExceptionWhenJobDeletedConcurrently` and
`testPullJobStatusSkipsJobDeletedConcurrently`.
All of `TestJobManager`, `TestJobMetaService`, and `TestJobPO` pass
locally.
---------
Co-authored-by: Claude Sonnet 5 <[email protected]>
---
.../tests/integration/test_model_catalog.py | 72 ++++--
.../tests/integration/test_relational_catalog.py | 4 +-
.../java/org/apache/gravitino/job/JobManager.java | 241 +++++++++++-------
.../gravitino/storage/relational/JDBCBackend.java | 2 +
.../storage/relational/mapper/JobMetaMapper.java | 3 +
.../mapper/JobMetaSQLProviderFactory.java | 5 +
.../provider/base/JobMetaBaseSQLProvider.java | 17 ++
.../gravitino/storage/relational/po/JobPO.java | 33 +++
.../storage/relational/service/JobMetaService.java | 75 +++++-
.../org/apache/gravitino/job/TestJobManager.java | 283 +++++++++++++++++++--
.../gravitino/storage/relational/po/TestJobPO.java | 61 +++++
.../relational/service/TestJobMetaService.java | 146 +++++++++++
12 files changed, 804 insertions(+), 138 deletions(-)
diff --git a/clients/client-python/tests/integration/test_model_catalog.py
b/clients/client-python/tests/integration/test_model_catalog.py
index 639288b74e..2bef34ef1e 100644
--- a/clients/client-python/tests/integration/test_model_catalog.py
+++ b/clients/client-python/tests/integration/test_model_catalog.py
@@ -255,7 +255,9 @@ class TestModelCatalog(IntegrationTestEnv):
self.assertEqual(update_property_model.name(), model_name)
self.assertEqual(update_property_model.comment(), comment)
self.assertEqual(update_property_model.latest_version(), 0)
- self.assert_properties_equal({"k1": "v11", "k2": "v2", "k3": "v3"},
update_property_model.properties())
+ self.assert_properties_equal(
+ {"k1": "v11", "k2": "v2", "k3": "v3"},
update_property_model.properties()
+ )
def test_register_remove_model_property(self):
model_name = f"model_it_model{str(randint(0, 1000))}"
@@ -327,7 +329,9 @@ class TestModelCatalog(IntegrationTestEnv):
self.assertEqual("uri", model_version.uri())
self.assertEqual(["alias1", "alias2"], model_version.aliases())
self.assertEqual("comment", model_version.comment())
- self.assert_properties_equal({"k1": "v1", "k2": "v2"},
model_version.properties())
+ self.assert_properties_equal(
+ {"k1": "v1", "k2": "v2"}, model_version.properties()
+ )
model_version =
self._catalog.as_model_catalog().get_model_version_by_alias(
model_ident, "alias1"
@@ -350,7 +354,9 @@ class TestModelCatalog(IntegrationTestEnv):
self.assertEqual(0, updated_model_version.version())
self.assertEqual("new comment", updated_model_version.comment())
self.assertEqual(["alias1", "alias2"], updated_model_version.aliases())
- self.assert_properties_equal({"k1": "v1", "k2": "v2"},
updated_model_version.properties())
+ self.assert_properties_equal(
+ {"k1": "v1", "k2": "v2"}, updated_model_version.properties()
+ )
self.assertEqual("uri", updated_model_version.uri())
def test_link_update_model_version_property(self):
@@ -379,7 +385,9 @@ class TestModelCatalog(IntegrationTestEnv):
self.assertEqual("uri", original_model_version.uri())
self.assertEqual(["alias1", "alias2"],
original_model_version.aliases())
self.assertEqual("comment", original_model_version.comment())
- self.assert_properties_equal({"k1": "v1", "k2": "v2"},
original_model_version.properties())
+ self.assert_properties_equal(
+ {"k1": "v1", "k2": "v2"}, original_model_version.properties()
+ )
changes = [
ModelVersionChange.set_property("k1", "v11"),
@@ -398,7 +406,9 @@ class TestModelCatalog(IntegrationTestEnv):
self.assertEqual(update_property_model.uri(), "uri")
self.assertEqual(update_property_model.comment(), comment)
self.assertEqual(update_property_model.aliases(), aliases)
- self.assert_properties_equal({"k1": "v11", "k3": "v3"},
update_property_model.properties())
+ self.assert_properties_equal(
+ {"k1": "v11", "k3": "v3"}, update_property_model.properties()
+ )
def test_link_update_model_version_uri(self):
model_name = f"model_it_model{str(randint(0, 1000))}"
@@ -425,7 +435,9 @@ class TestModelCatalog(IntegrationTestEnv):
self.assertEqual("uri", original_model_version.uri())
self.assertEqual(["alias1", "alias2"],
original_model_version.aliases())
self.assertEqual("comment", original_model_version.comment())
- self.assert_properties_equal({"k1": "v1", "k2": "v2"},
original_model_version.properties())
+ self.assert_properties_equal(
+ {"k1": "v1", "k2": "v2"}, original_model_version.properties()
+ )
changes = [ModelVersionChange.update_uri("new_uri")]
self._catalog.as_model_catalog().alter_model_version(model_ident, 0,
*changes)
@@ -437,7 +449,9 @@ class TestModelCatalog(IntegrationTestEnv):
self.assertEqual("new_uri", updated_model_version.uri())
self.assertEqual(["alias1", "alias2"], updated_model_version.aliases())
self.assertEqual("comment", updated_model_version.comment())
- self.assert_properties_equal({"k1": "v1", "k2": "v2"},
updated_model_version.properties())
+ self.assert_properties_equal(
+ {"k1": "v1", "k2": "v2"}, updated_model_version.properties()
+ )
def test_link_add_model_version_uri(self):
model_name = f"model_it_model{str(randint(0, 1000))}"
@@ -464,7 +478,9 @@ class TestModelCatalog(IntegrationTestEnv):
self.assertEqual({"n1": "u1"}, original_model_version.uris())
self.assertEqual(["alias1", "alias2"],
original_model_version.aliases())
self.assertEqual("comment", original_model_version.comment())
- self.assert_properties_equal({"k1": "v1", "k2": "v2"},
original_model_version.properties())
+ self.assert_properties_equal(
+ {"k1": "v1", "k2": "v2"}, original_model_version.properties()
+ )
changes = [ModelVersionChange.add_uri("n2", "u2")]
self._catalog.as_model_catalog().alter_model_version(model_ident, 0,
*changes)
@@ -476,7 +492,9 @@ class TestModelCatalog(IntegrationTestEnv):
self.assertEqual({"n1": "u1", "n2": "u2"},
updated_model_version.uris())
self.assertEqual(["alias1", "alias2"], updated_model_version.aliases())
self.assertEqual("comment", updated_model_version.comment())
- self.assert_properties_equal({"k1": "v1", "k2": "v2"},
updated_model_version.properties())
+ self.assert_properties_equal(
+ {"k1": "v1", "k2": "v2"}, updated_model_version.properties()
+ )
def test_link_remove_model_version_uri(self):
model_name = f"model_it_model{str(randint(0, 1000))}"
@@ -503,7 +521,9 @@ class TestModelCatalog(IntegrationTestEnv):
self.assertEqual({"n1": "u1", "n2": "u2"},
original_model_version.uris())
self.assertEqual(["alias1", "alias2"],
original_model_version.aliases())
self.assertEqual("comment", original_model_version.comment())
- self.assert_properties_equal({"k1": "v1", "k2": "v2"},
original_model_version.properties())
+ self.assert_properties_equal(
+ {"k1": "v1", "k2": "v2"}, original_model_version.properties()
+ )
changes = [ModelVersionChange.remove_uri("n1")]
self._catalog.as_model_catalog().alter_model_version(model_ident, 0,
*changes)
@@ -515,7 +535,9 @@ class TestModelCatalog(IntegrationTestEnv):
self.assertEqual({"n2": "u2"}, updated_model_version.uris())
self.assertEqual(["alias1", "alias2"], updated_model_version.aliases())
self.assertEqual("comment", updated_model_version.comment())
- self.assert_properties_equal({"k1": "v1", "k2": "v2"},
updated_model_version.properties())
+ self.assert_properties_equal(
+ {"k1": "v1", "k2": "v2"}, updated_model_version.properties()
+ )
def test_link_update_model_version_aliases(self):
model_name = f"model_it_model{str(randint(0, 1000))}"
@@ -542,7 +564,9 @@ class TestModelCatalog(IntegrationTestEnv):
self.assertEqual("uri", original_model_version.uri())
self.assertEqual(["alias1", "alias2"],
original_model_version.aliases())
self.assertEqual("comment", original_model_version.comment())
- self.assert_properties_equal({"k1": "v1", "k2": "v2"},
original_model_version.properties())
+ self.assert_properties_equal(
+ {"k1": "v1", "k2": "v2"}, original_model_version.properties()
+ )
# todo
changes = [
@@ -560,7 +584,9 @@ class TestModelCatalog(IntegrationTestEnv):
self.assertEqual("uri", updated_model_version.uri())
self.assertEqual(["alias2", "alias3"], updated_model_version.aliases())
self.assertEqual("comment", updated_model_version.comment())
- self.assert_properties_equal({"k1": "v1", "k2": "v2"},
updated_model_version.properties())
+ self.assert_properties_equal(
+ {"k1": "v1", "k2": "v2"}, updated_model_version.properties()
+ )
def test_link_update_model_version_aliases_from_empty(self):
# Regression test for https://github.com/apache/gravitino/issues/9727:
@@ -647,7 +673,9 @@ class TestModelCatalog(IntegrationTestEnv):
self.assertEqual("uri", model_version.uri())
self.assertEqual(["alias1", "alias2"], model_version.aliases())
self.assertEqual("comment", model_version.comment())
- self.assert_properties_equal({"k1": "v1", "k2": "v2"},
model_version.properties())
+ self.assert_properties_equal(
+ {"k1": "v1", "k2": "v2"}, model_version.properties()
+ )
model_version =
self._catalog.as_model_catalog().get_model_version_by_alias(
model_ident, "alias1"
@@ -768,7 +796,9 @@ class TestModelCatalog(IntegrationTestEnv):
self.assertEqual(model_versions[0].uri(), "uri1")
self.assertEqual(model_versions[0].comment(), "comment")
self.assertEqual(model_versions[0].aliases(), ["alias1", "alias2"])
- self.assert_properties_equal({"k1": "v1", "k2": "v2"},
model_versions[0].properties())
+ self.assert_properties_equal(
+ {"k1": "v1", "k2": "v2"}, model_versions[0].properties()
+ )
self.assertTrue(
self._catalog.as_model_catalog().delete_model_version(model_ident,
0)
@@ -805,7 +835,9 @@ class TestModelCatalog(IntegrationTestEnv):
self.assertEqual({"n1": "u1", "n2": "u2"}, model_version.uris())
self.assertEqual(["alias1", "alias2"], model_version.aliases())
self.assertEqual("comment", model_version.comment())
- self.assert_properties_equal({"k1": "v1", "k2": "v2"},
model_version.properties())
+ self.assert_properties_equal(
+ {"k1": "v1", "k2": "v2"}, model_version.properties()
+ )
# Test get model version by alias
model_version =
self._catalog.as_model_catalog().get_model_version_by_alias(
@@ -841,7 +873,9 @@ class TestModelCatalog(IntegrationTestEnv):
self.assertEqual({"n1": "u1", "n2": "u2"}, model_versions[0].uris())
self.assertEqual("comment", model_versions[0].comment())
self.assertEqual(["alias1", "alias2"], model_versions[0].aliases())
- self.assert_properties_equal({"k1": "v1", "k2": "v2"},
model_versions[0].properties())
+ self.assert_properties_equal(
+ {"k1": "v1", "k2": "v2"}, model_versions[0].properties()
+ )
def test_get_model_version_uri(self):
model_name = "model_it_model" + str(randint(0, 1000))
@@ -865,7 +899,9 @@ class TestModelCatalog(IntegrationTestEnv):
self.assertEqual({"n1": "u1", "n2": "u2"}, model_version.uris())
self.assertEqual(["alias1", "alias2"], model_version.aliases())
self.assertEqual("comment", model_version.comment())
- self.assert_properties_equal({"k1": "v1", "k2": "v2"},
model_version.properties())
+ self.assert_properties_equal(
+ {"k1": "v1", "k2": "v2"}, model_version.properties()
+ )
# Test get model version uri
model_version_uri =
self._catalog.as_model_catalog().get_model_version_uri(
diff --git a/clients/client-python/tests/integration/test_relational_catalog.py
b/clients/client-python/tests/integration/test_relational_catalog.py
index f336843f9d..dae7fa8f04 100644
--- a/clients/client-python/tests/integration/test_relational_catalog.py
+++ b/clients/client-python/tests/integration/test_relational_catalog.py
@@ -178,7 +178,9 @@ class TestRelationalCatalog(IntegrationTestEnv):
self.assertIsNotNone(table)
self.assertEqual(table.name(), TestRelationalCatalog.TABLE_NAME)
self.assertEqual(table.comment(), TestRelationalCatalog.TABLE_COMMENT)
- self.assert_properties_equal(TestRelationalCatalog.TABLE_PROPERTIES,
table.properties())
+ self.assert_properties_equal(
+ TestRelationalCatalog.TABLE_PROPERTIES, table.properties()
+ )
self.assertEqual(len(table.columns()), 3)
columns = table.columns()
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 aacb39a370..42df607c5e 100644
--- a/core/src/main/java/org/apache/gravitino/job/JobManager.java
+++ b/core/src/main/java/org/apache/gravitino/job/JobManager.java
@@ -509,34 +509,21 @@ public class JobManager implements JobOperationDispatcher
{
LockType.WRITE,
() -> {
try {
- // Re-fetch under the lock rather than reusing the snapshot taken
before the
- // (potentially slow) external cancel call above - a concurrent
status poll could
- // have persisted a real startedAt/finishedAt in that gap, and
carrying forward the
- // stale snapshot would clobber it back to the sentinel.
- JobEntity latestJobEntity = getJob(metalake, jobId);
- JobEntity newJobEntity =
- JobEntity.builder()
- .withId(latestJobEntity.id())
- .withJobExecutionId(latestJobEntity.jobExecutionId())
- .withJobTemplateName(latestJobEntity.jobTemplateName())
- .withStatus(JobHandle.Status.CANCELLING)
- .withNamespace(latestJobEntity.namespace())
- .withAuditInfo(
- AuditInfo.builder()
- .withCreator(latestJobEntity.auditInfo().creator())
-
.withCreateTime(latestJobEntity.auditInfo().createTime())
-
.withLastModifier(PrincipalUtils.getCurrentPrincipal().getName())
- .withLastModifiedTime(Instant.now())
- .build())
- // CANCELLING is not a terminal state; carry forward
whatever
- // startedAt/finishedAt the job already had.
- .withStartedAt(latestJobEntity.startedAt())
- .withFinishedAt(latestJobEntity.finishedAt())
- .build();
-
- // Update the job entity in the entity store
- entityStore.put(newJobEntity, true /* overwrite */);
- return newJobEntity;
+ // entityStore.update() re-fetches the latest entity itself right
before applying the
+ // updater, rather than reusing the snapshot taken before the
(potentially slow)
+ // external cancel call above - a concurrent status poll could
have persisted a real
+ // startedAt/finishedAt in that gap, and carrying forward the
stale snapshot would
+ // clobber it back to the sentinel.
+ return entityStore.update(
+ NameIdentifierUtil.ofJob(metalake, jobId),
+ JobEntity.class,
+ Entity.EntityType.JOB,
+ this::toCancellingJobEntity);
+ } catch (NoSuchEntityException e) {
+ throw new NoSuchJobException(
+ "Job with ID %s under metalake %s does not exist, this could
be due to the job "
+ + "not existing or being deleted concurrently.",
+ jobId, metalake);
} catch (IOException e) {
throw new RuntimeException(
String.format("Failed to update job entity for job %s to
CANCELING status", jobId),
@@ -545,6 +532,36 @@ public class JobManager implements JobOperationDispatcher {
});
}
+ private JobEntity toCancellingJobEntity(JobEntity latestJobEntity) {
+ // The external cancel call happens before this locked update, so a
concurrent status poll
+ // can persist a terminal status (or another cancelJob() call can already
have moved the job
+ // to CANCELLING) in the gap between the pre-cancel snapshot and this
re-fetch. Never regress
+ // the latest entity out of a terminal state, or overwrite an
already-CANCELLING one.
+ if (isFinishedStatus(latestJobEntity.status())
+ || latestJobEntity.status() == JobHandle.Status.CANCELLING) {
+ return latestJobEntity;
+ }
+
+ return JobEntity.builder()
+ .withId(latestJobEntity.id())
+ .withJobExecutionId(latestJobEntity.jobExecutionId())
+ .withJobTemplateName(latestJobEntity.jobTemplateName())
+ .withStatus(JobHandle.Status.CANCELLING)
+ .withNamespace(latestJobEntity.namespace())
+ .withAuditInfo(
+ AuditInfo.builder()
+ .withCreator(latestJobEntity.auditInfo().creator())
+ .withCreateTime(latestJobEntity.auditInfo().createTime())
+
.withLastModifier(PrincipalUtils.getCurrentPrincipal().getName())
+ .withLastModifiedTime(Instant.now())
+ .build())
+ // CANCELLING is not a terminal state; carry forward whatever
startedAt/finishedAt the
+ // job already had.
+ .withStartedAt(latestJobEntity.startedAt())
+ .withFinishedAt(latestJobEntity.finishedAt())
+ .build();
+ }
+
@Override
public void close() throws IOException {
try {
@@ -601,76 +618,126 @@ public class JobManager implements
JobOperationDispatcher {
}
if (newStatus != job.status()) {
- boolean isStarted = newStatus == JobHandle.Status.STARTED;
- boolean isFinished =
- newStatus == JobHandle.Status.SUCCEEDED
- || newStatus == JobHandle.Status.FAILED
- || newStatus == JobHandle.Status.CANCELLED;
-
- // Only a directly-observed STARTED transition is trustworthy
evidence of when a
- // job started. SUCCEEDED/FAILED do not prove the job ever
reached STARTED: FAILED
- // in particular can be reached directly from QUEUED (e.g.
NoSuchJobException from
- // the executor, or LocalJobExecutor failing before it records
STARTED), and even
- // for SUCCEEDED, backfilling startedAt from the queued time
would understate queue
- // latency and overstate execution duration in any derived
metric. So startedAt is
- // left unset unless a STARTED transition was actually observed.
- //
- // Only stamp startedAt on the first STARTED observation
(job.startedAt() <= 0).
- // A CANCELLING job already carries forward a real startedAt
from cancelJob, and
- // since cancellation is asynchronous, a poll can still observe
STARTED while
- // cancellation is in flight - overwriting the recorded start
time with this later
- // poll timestamp would lose the accurate value.
- long startedAt =
- isStarted && job.startedAt() <= 0
- ? Instant.now().toEpochMilli()
- : job.startedAt();
-
- JobEntity newJobEntity =
- JobEntity.builder()
- .withId(job.id())
- .withJobExecutionId(job.jobExecutionId())
- .withJobTemplateName(job.jobTemplateName())
- .withStatus(newStatus)
- .withNamespace(job.namespace())
- .withAuditInfo(
- AuditInfo.builder()
- .withCreator(job.auditInfo().creator())
- .withCreateTime(job.auditInfo().createTime())
-
.withLastModifier(PrincipalUtils.getCurrentPrincipal().getName())
- .withLastModifiedTime(Instant.now())
- .build())
- .withStartedAt(startedAt)
- .withFinishedAt(isFinished ?
Instant.now().toEpochMilli() : job.finishedAt())
- .build();
-
- // Update the job entity with new status.
+ // Update the job entity with new status. entityStore.update()
re-fetches the
+ // latest entity itself right before applying the updater, so
the transition below
+ // is derived from latestJobEntity - the state as of right
before the write - rather
+ // than the possibly-stale `job` snapshot taken by listJobs()
above. A concurrent
+ // writer (e.g. cancelJob(), or another poll run) may have
already moved the job to
+ // a terminal state, into CANCELLING, or recorded a real
startedAt/finishedAt in the
+ // gap between that snapshot and this point; the updater must
not regress any of
+ // that using the stale snapshot's view of the world.
JobHandle.Status finalNewStatus = newStatus;
- TreeLockUtils.doWithTreeLock(
- NameIdentifierUtil.ofJob(metalake, job.name()),
- LockType.WRITE,
- () -> {
- try {
- entityStore.put(newJobEntity, true /* overwrite */);
- return null;
- } catch (IOException e) {
- throw new RuntimeException(
- String.format(
- "Failed to update job entity %s to status %s",
- newJobEntity, finalNewStatus),
- e);
- }
- });
+ JobEntity updated;
+ try {
+ updated =
+ TreeLockUtils.doWithTreeLock(
+ NameIdentifierUtil.ofJob(metalake, job.name()),
+ LockType.WRITE,
+ () -> {
+ try {
+ return entityStore.update(
+ NameIdentifierUtil.ofJob(metalake, job.name()),
+ JobEntity.class,
+ Entity.EntityType.JOB,
+ latestJobEntity ->
+ toUpdatedStatusJobEntity(latestJobEntity,
finalNewStatus));
+ } catch (IOException e) {
+ throw new RuntimeException(
+ String.format(
+ "Failed to update job entity %s to status
%s",
+ job.name(), finalNewStatus),
+ e);
+ }
+ });
+ } 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
+ // than letting the exception escape this scheduled task,
which would silently
+ // cancel all future status-pull runs
(ScheduledExecutorService semantics).
+ LOG.warn(
+ "Job {} under metalake {} no longer exists, skipping
status update to {}. "
+ + "This could be due to the job being deleted
concurrently.",
+ job.name(),
+ metalake,
+ finalNewStatus);
+ return;
+ }
LOG.info(
"Updated the job {} with execution id {} status to {}",
job.name(),
job.jobExecutionId(),
- newStatus);
+ updated.status());
}
});
}
}
+ private JobEntity toUpdatedStatusJobEntity(
+ JobEntity latestJobEntity, JobHandle.Status observedStatus) {
+ JobHandle.Status currentStatus = latestJobEntity.status();
+ boolean observedIsFinished = isFinishedStatus(observedStatus);
+
+ // Never regress a job out of a terminal state, and never move a
CANCELLING job back to a
+ // non-terminal state - both would only be possible here because the
executor status was
+ // observed against a stale snapshot of the job.
+ if (isFinishedStatus(currentStatus)
+ || (currentStatus == JobHandle.Status.CANCELLING &&
!observedIsFinished)) {
+ return latestJobEntity;
+ }
+
+ // Only a directly-observed STARTED transition is trustworthy evidence of
when a job started.
+ // SUCCEEDED/FAILED do not prove the job ever reached STARTED: FAILED in
particular can be
+ // reached directly from QUEUED (e.g. NoSuchJobException from the
executor, or
+ // LocalJobExecutor failing before it records STARTED), and even for
SUCCEEDED, backfilling
+ // startedAt from the queued time would understate queue latency and
overstate execution
+ // duration in any derived metric. So startedAt is left unset unless a
STARTED transition was
+ // actually observed.
+ //
+ // Only stamp startedAt on the first STARTED observation
(latestJobEntity.startedAt() <= 0).
+ // A CANCELLING job already carries forward a real startedAt from
cancelJob, and since
+ // cancellation is asynchronous, a poll can still observe STARTED while
cancellation is in
+ // flight - overwriting the recorded start time with this later poll
timestamp would lose the
+ // accurate value.
+ boolean isStarted = observedStatus == JobHandle.Status.STARTED;
+ long startedAt =
+ isStarted && latestJobEntity.startedAt() <= 0
+ ? Instant.now().toEpochMilli()
+ : latestJobEntity.startedAt();
+
+ // Preserve an already-recorded finishedAt (e.g. stamped by a concurrent
writer) instead of
+ // overwriting it with a later poll's timestamp.
+ long finishedAt =
+ observedIsFinished
+ ? (latestJobEntity.finishedAt() > 0
+ ? latestJobEntity.finishedAt()
+ : Instant.now().toEpochMilli())
+ : latestJobEntity.finishedAt();
+
+ return JobEntity.builder()
+ .withId(latestJobEntity.id())
+ .withJobExecutionId(latestJobEntity.jobExecutionId())
+ .withJobTemplateName(latestJobEntity.jobTemplateName())
+ .withStatus(observedStatus)
+ .withNamespace(latestJobEntity.namespace())
+ .withAuditInfo(
+ AuditInfo.builder()
+ .withCreator(latestJobEntity.auditInfo().creator())
+ .withCreateTime(latestJobEntity.auditInfo().createTime())
+
.withLastModifier(PrincipalUtils.getCurrentPrincipal().getName())
+ .withLastModifiedTime(Instant.now())
+ .build())
+ .withStartedAt(startedAt)
+ .withFinishedAt(finishedAt)
+ .build();
+ }
+
+ private static boolean isFinishedStatus(JobHandle.Status status) {
+ return status == JobHandle.Status.SUCCEEDED
+ || status == JobHandle.Status.FAILED
+ || status == JobHandle.Status.CANCELLED;
+ }
+
@VisibleForTesting
void cleanUpStagingDirs() {
List<String> metalakes = MetalakeManager.listInUseMetalakes(entityStore);
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/JDBCBackend.java
b/core/src/main/java/org/apache/gravitino/storage/relational/JDBCBackend.java
index a0f3701d2e..11065934aa 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/JDBCBackend.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/JDBCBackend.java
@@ -1082,6 +1082,8 @@ public class JDBCBackend implements RelationalBackend,
SupportsOrphanedRelationC
return (E) PolicyMetaService.getInstance().updatePolicy(ident,
updater);
case JOB_TEMPLATE:
return (E)
JobTemplateMetaService.getInstance().updateJobTemplate(ident, updater);
+ case JOB:
+ return (E) JobMetaService.getInstance().updateJob(ident, updater);
case VIEW:
return (E) ViewMetaService.getInstance().updateView(ident, updater);
default:
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 7bfe738595..ae667fcad1 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
@@ -50,6 +50,9 @@ public interface JobMetaMapper {
JobPO selectJobPOByMetalakeAndRunId(
@Param("metalakeName") String metalakeName, @Param("jobRunId") Long
jobRunId);
+ @UpdateProvider(type = JobMetaSQLProviderFactory.class, method =
"updateJobMeta")
+ Integer updateJobMeta(@Param("newJobMeta") JobPO newJobPO,
@Param("oldJobMeta") JobPO oldJobPO);
+
@UpdateProvider(
type = JobMetaSQLProviderFactory.class,
method = "softDeleteJobMetaByMetalakeAndTemplate")
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 269e652504..20d6c414e1 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
@@ -76,6 +76,11 @@ public class JobMetaSQLProviderFactory {
return getProvider().selectJobPOByMetalakeAndRunId(metalakeName, jobRunId);
}
+ public static String updateJobMeta(
+ @Param("newJobMeta") JobPO newJobPO, @Param("oldJobMeta") JobPO
oldJobPO) {
+ return getProvider().updateJobMeta(newJobPO, oldJobPO);
+ }
+
public static String softDeleteJobMetaByMetalakeAndTemplate(
@Param("metalakeName") String metalakeName,
@Param("jobTemplateName") String jobTemplateName) {
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 5babf351fd..a18d45e73b 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
@@ -138,6 +138,23 @@ public class JobMetaBaseSQLProvider {
+ " AND jrm.deleted_at = 0 AND mm.deleted_at = 0 AND jtm.deleted_at =
0";
}
+ public String updateJobMeta(
+ @Param("newJobMeta") JobPO newJobPO, @Param("oldJobMeta") JobPO
oldJobPO) {
+ return "UPDATE "
+ + JobMetaMapper.TABLE_NAME
+ + " SET job_execution_id = #{newJobMeta.jobExecutionId},"
+ + " job_run_status = #{newJobMeta.jobRunStatus},"
+ + " job_started_at = #{newJobMeta.jobStartedAt},"
+ + " job_finished_at = #{newJobMeta.jobFinishedAt},"
+ + " audit_info = #{newJobMeta.auditInfo},"
+ + " current_version = #{newJobMeta.currentVersion},"
+ + " 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) {
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/po/JobPO.java
b/core/src/main/java/org/apache/gravitino/storage/relational/po/JobPO.java
index c5db26071e..bf0dbdf41d 100644
--- a/core/src/main/java/org/apache/gravitino/storage/relational/po/JobPO.java
+++ b/core/src/main/java/org/apache/gravitino/storage/relational/po/JobPO.java
@@ -126,6 +126,39 @@ public class JobPO {
}
}
+ /**
+ * Builds the {@link JobPO} to persist for an update, carrying forward the
identity fields ({@code
+ * jobRunId}, {@code jobTemplateName}) from the old PO since a job's
template is immutable, and
+ * bumping the version counters. This does not perform any
optimistic-concurrency check; the
+ * caller is responsible for any such guarantee.
+ *
+ * @param oldJobPO the existing {@link JobPO} being updated
+ * @param newJobEntity the {@link JobEntity} with the updated
status/timestamps/audit info
+ * @param builder the builder to populate, pre-configured with the {@code
metalakeId}
+ * @return the {@code JobPO} object with updated fields
+ */
+ public static JobPO updateJobPO(JobPO oldJobPO, JobEntity newJobEntity,
JobPOBuilder builder) {
+ try {
+ Long lastVersion = oldJobPO.lastVersion() + 1;
+ Long currentVersion = lastVersion;
+
+ return builder
+ .withJobRunId(oldJobPO.jobRunId())
+ .withJobTemplateName(oldJobPO.jobTemplateName())
+ .withJobExecutionId(newJobEntity.jobExecutionId())
+ .withJobRunStatus(newJobEntity.status().name())
+ .withJobStartedAt(newJobEntity.startedAt())
+ .withJobFinishedAt(newJobEntity.finishedAt())
+
.withAuditInfo(JsonUtils.anyFieldMapper().writeValueAsString(newJobEntity.auditInfo()))
+ .withCurrentVersion(currentVersion)
+ .withLastVersion(lastVersion)
+ .withDeletedAt(DEFAULT_DELETED_AT)
+ .build();
+ } catch (JsonProcessingException e) {
+ throw new RuntimeException("Failed to serialize job entity", e);
+ }
+ }
+
public static JobEntity fromJobPO(JobPO jobPO, Namespace namespace) {
try {
return JobEntity.builder()
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 85b33548ec..a67b8a21f3 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
@@ -20,11 +20,15 @@ package org.apache.gravitino.storage.relational.service;
import static
org.apache.gravitino.metrics.source.MetricsSource.GRAVITINO_RELATIONAL_STORE_METRIC_NAME;
+import com.google.common.base.Preconditions;
import java.io.IOException;
import java.util.List;
import java.util.Locale;
+import java.util.Objects;
+import java.util.function.Function;
import java.util.stream.Collectors;
import org.apache.gravitino.Entity;
+import org.apache.gravitino.HasIdentifier;
import org.apache.gravitino.NameIdentifier;
import org.apache.gravitino.Namespace;
import org.apache.gravitino.exceptions.IllegalNamespaceException;
@@ -84,19 +88,7 @@ public class JobMetaService {
metricsSource = GRAVITINO_RELATIONAL_STORE_METRIC_NAME,
baseMetricName = "getJobByIdentifier")
public JobEntity getJobByIdentifier(NameIdentifier ident) {
- String metalakeName = ident.namespace().level(0);
- long jobRunIdLong = parseJobRunId(ident.name());
-
- JobPO jobPO =
- SessionUtils.getWithoutCommit(
- JobMetaMapper.class,
- mapper -> mapper.selectJobPOByMetalakeAndRunId(metalakeName,
jobRunIdLong));
- if (jobPO == null) {
- throw new NoSuchEntityException(
- NoSuchEntityException.NO_SUCH_ENTITY_MESSAGE,
- Entity.EntityType.JOB.name().toLowerCase(Locale.ROOT),
- ident.toString());
- }
+ JobPO jobPO = getJobPO(ident);
return JobPO.fromJobPO(jobPO, ident.namespace());
}
@@ -125,6 +117,46 @@ public class JobMetaService {
}
}
+ @Monitored(metricsSource = GRAVITINO_RELATIONAL_STORE_METRIC_NAME,
baseMetricName = "updateJob")
+ public <E extends Entity & HasIdentifier> JobEntity updateJob(
+ NameIdentifier jobIdent, Function<E, E> updater) throws IOException {
+ JobPO oldJobPO = getJobPO(jobIdent);
+ JobEntity oldJobEntity = JobPO.fromJobPO(oldJobPO, jobIdent.namespace());
+ JobEntity newJobEntity = (JobEntity) updater.apply((E) oldJobEntity);
+ Preconditions.checkArgument(
+ Objects.equals(oldJobEntity.id(), newJobEntity.id()),
+ "The updated job entity id: %s is not equal to the old one: %s, which
is unexpected",
+ newJobEntity.id(),
+ oldJobEntity.id());
+
+ 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));
+ } catch (RuntimeException e) {
+ ExceptionUtils.checkSQLException(e, Entity.EntityType.JOB,
oldJobEntity.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;
+ }
+ }
+
@Monitored(metricsSource = GRAVITINO_RELATIONAL_STORE_METRIC_NAME,
baseMetricName = "deleteJob")
public boolean deleteJob(NameIdentifier jobIdent) {
long jobRunIdLong = parseJobRunId(jobIdent.name());
@@ -147,6 +179,23 @@ public class JobMetaService {
mapper -> mapper.deleteJobMetasByLegacyTimeline(legacyTimeline,
limit));
}
+ private JobPO getJobPO(NameIdentifier ident) {
+ String metalakeName = ident.namespace().level(0);
+ long jobRunIdLong = parseJobRunId(ident.name());
+
+ JobPO jobPO =
+ SessionUtils.getWithoutCommit(
+ JobMetaMapper.class,
+ mapper -> mapper.selectJobPOByMetalakeAndRunId(metalakeName,
jobRunIdLong));
+ if (jobPO == null) {
+ throw new NoSuchEntityException(
+ NoSuchEntityException.NO_SUCH_ENTITY_MESSAGE,
+ Entity.EntityType.JOB.name().toLowerCase(Locale.ROOT),
+ ident.toString());
+ }
+ return jobPO;
+ }
+
// Validate and parse a job run identifier of the form "job-<number>";
// throws NoSuchEntityException for any malformed input instead of leaking
parsing errors.
private long parseJobRunId(String jobRunId) {
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 9b5ec6e27b..c1345d0f77 100644
--- a/core/src/test/java/org/apache/gravitino/job/TestJobManager.java
+++ b/core/src/test/java/org/apache/gravitino/job/TestJobManager.java
@@ -20,6 +20,7 @@ package org.apache.gravitino.job;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyBoolean;
+import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.doNothing;
import static org.mockito.Mockito.doReturn;
import static org.mockito.Mockito.doThrow;
@@ -47,6 +48,7 @@ import java.util.Random;
import java.util.UUID;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
+import java.util.function.Function;
import org.apache.commons.io.FileUtils;
import org.apache.commons.lang3.ArrayUtils;
import org.apache.commons.lang3.reflect.FieldUtils;
@@ -561,7 +563,7 @@ public class TestJobManager {
JobEntity job = newJobEntity("shell_job", JobHandle.Status.QUEUED);
when(jobManager.getJob(metalake, job.name())).thenReturn(job);
doNothing().when(jobExecutor).cancelJob(job.jobExecutionId());
- doNothing().when(entityStore).put(any(JobEntity.class), anyBoolean());
+ stubEntityStoreUpdateToApply(job);
// Cancel an existing job
JobEntity cancelledJob = jobManager.cancelJob(metalake, job.name());
@@ -601,12 +603,95 @@ public class TestJobManager {
// Test when entity store failed to update the job status
doThrow(new IOException("Entity store error"))
.when(entityStore)
- .put(any(JobEntity.class), anyBoolean());
+ .update(any(), eq(JobEntity.class), eq(Entity.EntityType.JOB), any());
Assertions.assertThrows(
RuntimeException.class, () -> jobManager.cancelJob(metalake,
job.name()));
}
+ @Test
+ public void
testCancelJobThrowsNoSuchJobExceptionWhenJobDeletedConcurrently() throws
IOException {
+ mockedMetalake
+ .when(() -> MetalakeManager.checkMetalake(metalakeIdent, entityStore))
+ .thenAnswer(a -> null);
+
+ JobEntity job = newJobEntity("shell_job", JobHandle.Status.QUEUED);
+ when(jobManager.getJob(metalake, job.name())).thenReturn(job);
+ doNothing().when(jobExecutor).cancelJob(job.jobExecutionId());
+
+ // Simulate the job having been deleted concurrently (e.g. by
legacy-timeline cleanup) in the
+ // gap between the getJob() snapshot above and the entityStore.update()
call.
+ when(entityStore.update(any(), eq(JobEntity.class),
eq(Entity.EntityType.JOB), any()))
+ .thenThrow(new NoSuchEntityException("Job does not exist"));
+
+ Assertions.assertThrows(
+ NoSuchJobException.class, () -> jobManager.cancelJob(metalake,
job.name()));
+ }
+
+ @Test
+ public void testCancelJobDoesNotRegressConcurrentlyFinishedJob() throws
IOException {
+ mockedMetalake
+ .when(() -> MetalakeManager.checkMetalake(metalakeIdent, entityStore))
+ .thenAnswer(a -> null);
+
+ // getJob() observes the job as QUEUED (active), so the external cancel
call fires. But by
+ // the time entityStore.update() re-fetches the entity, a concurrent
status poll has already
+ // persisted a terminal status - that must not be regressed back to
CANCELLING.
+ JobEntity queuedSnapshot = newJobEntity("shell_job",
JobHandle.Status.QUEUED);
+ when(jobManager.getJob(metalake,
queuedSnapshot.name())).thenReturn(queuedSnapshot);
+ doNothing().when(jobExecutor).cancelJob(queuedSnapshot.jobExecutionId());
+
+ JobEntity latestSucceeded =
+ JobEntity.builder()
+ .withId(queuedSnapshot.id())
+ .withJobExecutionId(queuedSnapshot.jobExecutionId())
+ .withNamespace(queuedSnapshot.namespace())
+ .withJobTemplateName(queuedSnapshot.jobTemplateName())
+ .withStatus(JobHandle.Status.SUCCEEDED)
+ .withAuditInfo(queuedSnapshot.auditInfo())
+ .withStartedAt(12345L)
+ .withFinishedAt(67890L)
+ .build();
+ stubEntityStoreUpdateToApply(latestSucceeded);
+
+ JobEntity result = jobManager.cancelJob(metalake, queuedSnapshot.name());
+ Assertions.assertEquals(JobHandle.Status.SUCCEEDED, result.status());
+ Assertions.assertEquals(12345L, result.startedAt());
+ Assertions.assertEquals(67890L, result.finishedAt());
+ }
+
+ @Test
+ public void testCancelJobDoesNotRegressAlreadyCancellingJob() throws
IOException {
+ mockedMetalake
+ .when(() -> MetalakeManager.checkMetalake(metalakeIdent, entityStore))
+ .thenAnswer(a -> null);
+
+ // A concurrent cancelJob() call already moved the job to CANCELLING
between the getJob()
+ // snapshot and this update; re-applying CANCELLING here must not stamp a
fresh
+ // lastModifiedTime over the entity the other writer already wrote.
+ JobEntity queuedSnapshot = newJobEntity("shell_job",
JobHandle.Status.QUEUED);
+ when(jobManager.getJob(metalake,
queuedSnapshot.name())).thenReturn(queuedSnapshot);
+ doNothing().when(jobExecutor).cancelJob(queuedSnapshot.jobExecutionId());
+
+ JobEntity latestCancelling =
+ JobEntity.builder()
+ .withId(queuedSnapshot.id())
+ .withJobExecutionId(queuedSnapshot.jobExecutionId())
+ .withNamespace(queuedSnapshot.namespace())
+ .withJobTemplateName(queuedSnapshot.jobTemplateName())
+ .withStatus(JobHandle.Status.CANCELLING)
+ .withAuditInfo(queuedSnapshot.auditInfo())
+ .withStartedAt(12345L)
+ .withFinishedAt(0L)
+ .build();
+ stubEntityStoreUpdateToApply(latestCancelling);
+
+ JobEntity result = jobManager.cancelJob(metalake, queuedSnapshot.name());
+ Assertions.assertEquals(JobHandle.Status.CANCELLING, result.status());
+ Assertions.assertEquals(12345L, result.startedAt());
+ Assertions.assertEquals(0L, result.finishedAt());
+ }
+
@Test
public void testPullJobStatus() throws IOException {
JobEntity job = newJobEntity("shell_job", JobHandle.Status.QUEUED);
@@ -629,16 +714,15 @@ public class TestJobManager {
when(jobExecutor.getJobStatus(job.jobExecutionId())).thenReturn(JobHandle.Status.QUEUED);
Assertions.assertDoesNotThrow(() -> jobManager.pullAndUpdateJobStatus());
- verify(entityStore, never()).put(any(), anyBoolean());
+ verify(entityStore, never())
+ .update(any(), eq(JobEntity.class), eq(Entity.EntityType.JOB), any());
+ stubEntityStoreUpdateToApply(job);
when(jobExecutor.getJobStatus(job.jobExecutionId())).thenReturn(JobHandle.Status.SUCCEEDED);
Assertions.assertDoesNotThrow(() -> jobManager.pullAndUpdateJobStatus());
- ArgumentCaptor<JobEntity> captor =
ArgumentCaptor.forClass(JobEntity.class);
- verify(entityStore, times(1)).put(captor.capture(), anyBoolean());
-
// Once a job transitions to a terminal status, finishedAt must be set.
- JobEntity updatedJob = captor.getValue();
+ JobEntity updatedJob = captureUpdatedJobEntity(job);
Assertions.assertEquals(JobHandle.Status.SUCCEEDED, updatedJob.status());
Assertions.assertNotNull(updatedJob.finishedAt());
Assertions.assertTrue(updatedJob.finishedAt() > 0);
@@ -675,12 +759,11 @@ public class TestJobManager {
when(jobManager.listJobs(metalake,
Optional.empty())).thenReturn(ImmutableList.of(job));
// QUEUED -> STARTED: startedAt must be set, finishedAt must remain unset.
+ stubEntityStoreUpdateToApply(job);
when(jobExecutor.getJobStatus(job.jobExecutionId())).thenReturn(JobHandle.Status.STARTED);
Assertions.assertDoesNotThrow(() -> jobManager.pullAndUpdateJobStatus());
- ArgumentCaptor<JobEntity> startedCaptor =
ArgumentCaptor.forClass(JobEntity.class);
- verify(entityStore, times(1)).put(startedCaptor.capture(), anyBoolean());
- JobEntity startedJob = startedCaptor.getValue();
+ JobEntity startedJob = captureUpdatedJobEntity(job);
Assertions.assertEquals(JobHandle.Status.STARTED, startedJob.status());
Assertions.assertNotNull(startedJob.startedAt());
Assertions.assertTrue(startedJob.startedAt() > 0);
@@ -689,13 +772,12 @@ public class TestJobManager {
// STARTED -> SUCCEEDED: finishedAt must be set, and the
previously-recorded startedAt must
// be carried forward unchanged, not overwritten.
Mockito.clearInvocations(entityStore);
+ stubEntityStoreUpdateToApply(startedJob);
when(jobManager.listJobs(metalake,
Optional.empty())).thenReturn(ImmutableList.of(startedJob));
when(jobExecutor.getJobStatus(job.jobExecutionId())).thenReturn(JobHandle.Status.SUCCEEDED);
Assertions.assertDoesNotThrow(() -> jobManager.pullAndUpdateJobStatus());
- ArgumentCaptor<JobEntity> finishedCaptor =
ArgumentCaptor.forClass(JobEntity.class);
- verify(entityStore, times(1)).put(finishedCaptor.capture(), anyBoolean());
- JobEntity finishedJob = finishedCaptor.getValue();
+ JobEntity finishedJob = captureUpdatedJobEntity(startedJob);
Assertions.assertEquals(JobHandle.Status.SUCCEEDED, finishedJob.status());
Assertions.assertEquals(startedJob.startedAt(), finishedJob.startedAt());
Assertions.assertNotNull(finishedJob.finishedAt());
@@ -737,13 +819,12 @@ public class TestJobManager {
.thenReturn(ImmutableList.of(metalake));
when(jobManager.listJobs(metalake,
Optional.empty())).thenReturn(ImmutableList.of(queuedJob));
+ stubEntityStoreUpdateToApply(queuedJob);
when(jobExecutor.getJobStatus(queuedJob.jobExecutionId()))
.thenReturn(JobHandle.Status.SUCCEEDED);
Assertions.assertDoesNotThrow(() -> jobManager.pullAndUpdateJobStatus());
- ArgumentCaptor<JobEntity> captor =
ArgumentCaptor.forClass(JobEntity.class);
- verify(entityStore, times(1)).put(captor.capture(), anyBoolean());
- JobEntity succeededJob = captor.getValue();
+ JobEntity succeededJob = captureUpdatedJobEntity(queuedJob);
Assertions.assertEquals(JobHandle.Status.SUCCEEDED, succeededJob.status());
Assertions.assertEquals(0L, succeededJob.startedAt());
Assertions.assertNotNull(succeededJob.finishedAt());
@@ -782,19 +863,161 @@ public class TestJobManager {
when(jobManager.listJobs(metalake, Optional.empty()))
.thenReturn(ImmutableList.of(cancellingJob));
+ stubEntityStoreUpdateToApply(cancellingJob);
when(jobExecutor.getJobStatus(cancellingJob.jobExecutionId()))
.thenReturn(JobHandle.Status.CANCELLED);
Assertions.assertDoesNotThrow(() -> jobManager.pullAndUpdateJobStatus());
- ArgumentCaptor<JobEntity> captor =
ArgumentCaptor.forClass(JobEntity.class);
- verify(entityStore, times(1)).put(captor.capture(), anyBoolean());
- JobEntity cancelledJob = captor.getValue();
+ JobEntity cancelledJob = captureUpdatedJobEntity(cancellingJob);
Assertions.assertEquals(JobHandle.Status.CANCELLED, cancelledJob.status());
Assertions.assertEquals(0L, cancelledJob.startedAt());
Assertions.assertNotNull(cancelledJob.finishedAt());
Assertions.assertTrue(cancelledJob.finishedAt() > 0);
}
+ @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());
+
+ 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());
+ }
+
+ @Test
+ public void testPullJobStatusDoesNotRegressConcurrentlyFinishedJob() throws
IOException {
+ // listJobs() observes the job as QUEUED, but by the time
entityStore.update() re-fetches it,
+ // a concurrent writer (e.g. another poll run) has already finished the
job with a different
+ // terminal status. The stale QUEUED snapshot - and the executor status
derived from it - must
+ // not be allowed to regress that terminal state or clobber its recorded
startedAt/finishedAt.
+ JobEntity queuedSnapshot = newJobEntity("shell_job",
JobHandle.Status.QUEUED);
+ JobEntity latestSucceeded =
+ JobEntity.builder()
+ .withId(queuedSnapshot.id())
+ .withJobExecutionId(queuedSnapshot.jobExecutionId())
+ .withNamespace(queuedSnapshot.namespace())
+ .withJobTemplateName(queuedSnapshot.jobTemplateName())
+ .withStatus(JobHandle.Status.SUCCEEDED)
+ .withAuditInfo(queuedSnapshot.auditInfo())
+ .withStartedAt(12345L)
+ .withFinishedAt(67890L)
+ .build();
+
+ 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(queuedSnapshot));
+ stubEntityStoreUpdateToApply(latestSucceeded);
+ // The stale QUEUED snapshot leads the poll to observe (and try to apply)
FAILED - a
+ // different terminal status than the one the job has actually already
settled into.
+ when(jobExecutor.getJobStatus(queuedSnapshot.jobExecutionId()))
+ .thenReturn(JobHandle.Status.FAILED);
+ Assertions.assertDoesNotThrow(() -> jobManager.pullAndUpdateJobStatus());
+
+ JobEntity result = captureUpdatedJobEntity(latestSucceeded);
+ Assertions.assertEquals(JobHandle.Status.SUCCEEDED, result.status());
+ Assertions.assertEquals(12345L, result.startedAt());
+ Assertions.assertEquals(67890L, result.finishedAt());
+ }
+
+ @Test
+ public void testPullJobStatusDoesNotRegressConcurrentlyCancellingJob()
throws IOException {
+ // listJobs() observes the job as QUEUED, but a concurrent cancelJob()
moves it to CANCELLING
+ // before entityStore.update() re-fetches it. The executor reports STARTED
for this poll
+ // (a legitimate observation for the same jobExecutionId) - that must not
move the job back
+ // out of CANCELLING, nor overwrite the startedAt it already carries.
+ JobEntity queuedSnapshot = newJobEntity("shell_job",
JobHandle.Status.QUEUED);
+ JobEntity latestCancelling =
+ JobEntity.builder()
+ .withId(queuedSnapshot.id())
+ .withJobExecutionId(queuedSnapshot.jobExecutionId())
+ .withNamespace(queuedSnapshot.namespace())
+ .withJobTemplateName(queuedSnapshot.jobTemplateName())
+ .withStatus(JobHandle.Status.CANCELLING)
+ .withAuditInfo(queuedSnapshot.auditInfo())
+ .withStartedAt(12345L)
+ .withFinishedAt(0L)
+ .build();
+
+ 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(queuedSnapshot));
+ stubEntityStoreUpdateToApply(latestCancelling);
+ when(jobExecutor.getJobStatus(queuedSnapshot.jobExecutionId()))
+ .thenReturn(JobHandle.Status.STARTED);
+ Assertions.assertDoesNotThrow(() -> jobManager.pullAndUpdateJobStatus());
+
+ JobEntity result = captureUpdatedJobEntity(latestCancelling);
+ Assertions.assertEquals(JobHandle.Status.CANCELLING, result.status());
+ Assertions.assertEquals(12345L, result.startedAt());
+ Assertions.assertEquals(0L, result.finishedAt());
+ }
+
@Test
public void testCleanUpStagingDirs() throws IOException,
InterruptedException {
JobEntity job = newJobEntity("shell_job", JobHandle.Status.STARTED);
@@ -1103,6 +1326,28 @@ public class TestJobManager {
.build();
}
+ // cancelJob/pullAndUpdateJobStatus now go through entityStore.update(),
which re-fetches the
+ // latest entity and applies an updater function internally. Since
entityStore is a full mock,
+ // this stubs that re-fetch to hand back the given entity, mirroring what
the real
+ // JobMetaService.updateJob would read from storage.
+ @SuppressWarnings("unchecked")
+ private void stubEntityStoreUpdateToApply(JobEntity latestJobEntity) throws
IOException {
+ when(entityStore.update(any(), eq(JobEntity.class),
eq(Entity.EntityType.JOB), any()))
+ .thenAnswer(
+ invocation -> {
+ Function<JobEntity, JobEntity> updater =
invocation.getArgument(3);
+ return updater.apply(latestJobEntity);
+ });
+ }
+
+ @SuppressWarnings("unchecked")
+ private JobEntity captureUpdatedJobEntity(JobEntity latestJobEntity) throws
IOException {
+ ArgumentCaptor<Function<JobEntity, JobEntity>> captor =
ArgumentCaptor.forClass(Function.class);
+ verify(entityStore, times(1))
+ .update(any(), eq(JobEntity.class), eq(Entity.EntityType.JOB),
captor.capture());
+ return captor.getValue().apply(latestJobEntity);
+ }
+
private HttpServer createLoopbackHttpServer(String response) throws
IOException {
HttpServer server = HttpServer.create(new InetSocketAddress("127.0.0.1",
0), 0);
server.createContext(
diff --git
a/core/src/test/java/org/apache/gravitino/storage/relational/po/TestJobPO.java
b/core/src/test/java/org/apache/gravitino/storage/relational/po/TestJobPO.java
index 76a390b173..9ad14ed737 100644
---
a/core/src/test/java/org/apache/gravitino/storage/relational/po/TestJobPO.java
+++
b/core/src/test/java/org/apache/gravitino/storage/relational/po/TestJobPO.java
@@ -200,4 +200,65 @@ public class TestJobPO {
Assertions.assertEquals(finishedAt, jobPO.jobFinishedAt());
Assertions.assertEquals(finishedAt, resultEntity.finishedAt());
}
+
+ @Test
+ public void testUpdateJobPO() {
+ JobEntity oldEntity =
+ JobEntity.builder()
+ .withId(1L)
+ .withJobExecutionId("job-execution-1")
+ .withJobTemplateName("test-job-template")
+ .withStatus(JobHandle.Status.QUEUED)
+ .withNamespace(NamespaceUtil.ofJob("test"))
+ .withAuditInfo(
+
AuditInfo.builder().withCreator("test").withCreateTime(Instant.now()).build())
+ .withStartedAt(0L)
+ .withFinishedAt(0L)
+ .build();
+
+ JobPO.JobPOBuilder initBuilder = JobPO.builder().withMetalakeId(1L);
+ JobPO oldJobPO = JobPO.initializeJobPO(oldEntity, initBuilder);
+
+ long startedAt = Instant.now().toEpochMilli();
+ AuditInfo updatedAuditInfo =
+ AuditInfo.builder()
+ .withCreator(oldEntity.auditInfo().creator())
+ .withCreateTime(oldEntity.auditInfo().createTime())
+ .withLastModifier("updater")
+ .withLastModifiedTime(Instant.now())
+ .build();
+
+ // Deliberately try to change jobTemplateName - updateJobPO must ignore it
and keep the old
+ // PO's value, since a job's template is immutable once the job is created.
+ JobEntity newEntity =
+ JobEntity.builder()
+ .withId(oldEntity.id())
+ .withJobExecutionId("job-execution-2")
+ .withJobTemplateName("a-different-job-template")
+ .withStatus(JobHandle.Status.STARTED)
+ .withNamespace(oldEntity.namespace())
+ .withAuditInfo(updatedAuditInfo)
+ .withStartedAt(startedAt)
+ .withFinishedAt(0L)
+ .build();
+
+ JobPO.JobPOBuilder updateBuilder =
JobPO.builder().withMetalakeId(oldJobPO.metalakeId());
+ JobPO newJobPO = JobPO.updateJobPO(oldJobPO, newEntity, updateBuilder);
+
+ Assertions.assertEquals(oldJobPO.jobRunId(), newJobPO.jobRunId());
+ Assertions.assertEquals(oldJobPO.jobTemplateName(),
newJobPO.jobTemplateName());
+ Assertions.assertEquals("job-execution-2", newJobPO.jobExecutionId());
+ Assertions.assertEquals(JobHandle.Status.STARTED.name(),
newJobPO.jobRunStatus());
+ Assertions.assertEquals(startedAt, newJobPO.jobStartedAt());
+ Assertions.assertEquals(0L, newJobPO.jobFinishedAt());
+ Assertions.assertEquals(oldJobPO.currentVersion() + 1,
newJobPO.currentVersion());
+ Assertions.assertEquals(oldJobPO.lastVersion() + 1,
newJobPO.lastVersion());
+
+ JobEntity resultEntity = JobPO.fromJobPO(newJobPO,
NamespaceUtil.ofJob("test"));
+ Assertions.assertEquals(JobHandle.Status.STARTED, resultEntity.status());
+ Assertions.assertEquals(startedAt, resultEntity.startedAt());
+ Assertions.assertEquals("test-job-template",
resultEntity.jobTemplateName());
+ Assertions.assertEquals(
+ updatedAuditInfo.lastModifier(),
resultEntity.auditInfo().lastModifier());
+ }
}
diff --git
a/core/src/test/java/org/apache/gravitino/storage/relational/service/TestJobMetaService.java
b/core/src/test/java/org/apache/gravitino/storage/relational/service/TestJobMetaService.java
index a45294e553..fd17917d93 100644
---
a/core/src/test/java/org/apache/gravitino/storage/relational/service/TestJobMetaService.java
+++
b/core/src/test/java/org/apache/gravitino/storage/relational/service/TestJobMetaService.java
@@ -232,6 +232,152 @@ public class TestJobMetaService extends TestJDBCBackend {
.deleteJob(NameIdentifierUtil.ofJob(METALAKE_NAME, job.name())));
}
+ @TestTemplate
+ public void testUpdateJob() throws IOException {
+ BaseMetalake metalake =
+ createBaseMakeLake(RandomIdGenerator.INSTANCE.nextId(), METALAKE_NAME,
AUDIT_INFO);
+ backend.insert(metalake, false);
+
+ JobTemplateEntity jobTemplate =
+ TestJobTemplateMetaService.newShellJobTemplateEntity(
+ "test_job_template", "test_comment", METALAKE_NAME);
+ JobTemplateMetaService.getInstance().insertJobTemplate(jobTemplate, false);
+
+ JobEntity job =
+ TestJobTemplateMetaService.newJobEntity(
+ jobTemplate.name(), JobHandle.Status.QUEUED, METALAKE_NAME);
+ JobMetaService.getInstance().insertJob(job, false);
+
+ // Update the job to STARTED, setting startedAt and a new audit info.
+ long startedAt = System.currentTimeMillis();
+ AuditInfo startedAuditInfo =
+ AuditInfo.builder()
+ .withCreator(job.auditInfo().creator())
+ .withCreateTime(job.auditInfo().createTime())
+ .withLastModifier("updater")
+ .withLastModifiedTime(Instant.now())
+ .build();
+
+ JobEntity startedJob =
+ JobMetaService.getInstance()
+ .updateJob(
+ NameIdentifierUtil.ofJob(METALAKE_NAME, job.name()),
+ (JobEntity oldJob) ->
+ JobEntity.builder()
+ .withId(oldJob.id())
+ .withJobExecutionId(oldJob.jobExecutionId())
+ .withJobTemplateName(oldJob.jobTemplateName())
+ .withNamespace(oldJob.namespace())
+ .withStatus(JobHandle.Status.STARTED)
+ .withAuditInfo(startedAuditInfo)
+ .withStartedAt(startedAt)
+ .withFinishedAt(oldJob.finishedAt())
+ .build());
+
+ Assertions.assertEquals(JobHandle.Status.STARTED, startedJob.status());
+ Assertions.assertEquals(startedAt, startedJob.startedAt());
+ Assertions.assertEquals(startedAuditInfo, startedJob.auditInfo());
+ Assertions.assertEquals(0L, startedJob.finishedAt());
+
+ // The update must actually be persisted, not just returned.
+ JobEntity fetchedJob =
+ JobMetaService.getInstance()
+ .getJobByIdentifier(NameIdentifierUtil.ofJob(METALAKE_NAME,
job.name()));
+ Assertions.assertEquals(startedJob, fetchedJob);
+
+ // A second, independent update must keep working (e.g. an internal
version counter bumped by
+ // the first update must not break a subsequent one).
+ long finishedAt = startedAt + 1000;
+ JobEntity finishedJob =
+ JobMetaService.getInstance()
+ .updateJob(
+ NameIdentifierUtil.ofJob(METALAKE_NAME, job.name()),
+ (JobEntity oldJob) ->
+ JobEntity.builder()
+ .withId(oldJob.id())
+ .withJobExecutionId(oldJob.jobExecutionId())
+ .withJobTemplateName(oldJob.jobTemplateName())
+ .withNamespace(oldJob.namespace())
+ .withStatus(JobHandle.Status.SUCCEEDED)
+ .withAuditInfo(oldJob.auditInfo())
+ .withStartedAt(oldJob.startedAt())
+ .withFinishedAt(finishedAt)
+ .build());
+
+ Assertions.assertEquals(JobHandle.Status.SUCCEEDED, finishedJob.status());
+ Assertions.assertEquals(startedAt, finishedJob.startedAt());
+ Assertions.assertEquals(finishedAt, finishedJob.finishedAt());
+
+ fetchedJob =
+ JobMetaService.getInstance()
+ .getJobByIdentifier(NameIdentifierUtil.ofJob(METALAKE_NAME,
job.name()));
+ Assertions.assertEquals(finishedJob, fetchedJob);
+ }
+
+ @TestTemplate
+ public void testUpdateNonExistentJobThrowsNoSuchEntityException() throws
IOException {
+ BaseMetalake metalake =
+ createBaseMakeLake(RandomIdGenerator.INSTANCE.nextId(), METALAKE_NAME,
AUDIT_INFO);
+ backend.insert(metalake, false);
+
+ Assertions.assertThrows(
+ NoSuchEntityException.class,
+ () ->
+ JobMetaService.getInstance()
+ .updateJob(NameIdentifierUtil.ofJob(METALAKE_NAME,
"job-999999"), e -> e));
+ }
+
+ @TestTemplate
+ public void testUpdateJobWithMismatchedIdThrowsIllegalArgumentException()
throws IOException {
+ BaseMetalake metalake =
+ createBaseMakeLake(RandomIdGenerator.INSTANCE.nextId(), METALAKE_NAME,
AUDIT_INFO);
+ backend.insert(metalake, false);
+
+ JobTemplateEntity jobTemplate =
+ TestJobTemplateMetaService.newShellJobTemplateEntity(
+ "test_job_template", "test_comment", METALAKE_NAME);
+ JobTemplateMetaService.getInstance().insertJobTemplate(jobTemplate, false);
+
+ JobEntity job =
+ TestJobTemplateMetaService.newJobEntity(
+ jobTemplate.name(), JobHandle.Status.QUEUED, METALAKE_NAME);
+ JobMetaService.getInstance().insertJob(job, false);
+
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ JobMetaService.getInstance()
+ .updateJob(
+ NameIdentifierUtil.ofJob(METALAKE_NAME, job.name()),
+ (JobEntity oldJob) ->
+ JobEntity.builder()
+ .withId(oldJob.id() + 1)
+ .withJobExecutionId(oldJob.jobExecutionId())
+ .withJobTemplateName(oldJob.jobTemplateName())
+ .withNamespace(oldJob.namespace())
+ .withStatus(oldJob.status())
+ .withAuditInfo(oldJob.auditInfo())
+ .withStartedAt(oldJob.startedAt())
+ .withFinishedAt(oldJob.finishedAt())
+ .build()));
+ }
+
+ @Test
+ public void
testUpdateJobWithMalformedIdentifierThrowsNoSuchEntityException() {
+ Assertions.assertThrows(
+ NoSuchEntityException.class,
+ () ->
+ JobMetaService.getInstance()
+ .updateJob(NameIdentifierUtil.ofJob(METALAKE_NAME, "invalid"),
e -> e));
+
+ Assertions.assertThrows(
+ NoSuchEntityException.class,
+ () ->
+ JobMetaService.getInstance()
+ .updateJob(
+ NameIdentifierUtil.ofJob(METALAKE_NAME,
JobHandle.JOB_ID_PREFIX), e -> e));
+ }
+
@Test
public void testGetJobWithMalformedIdentifierThrowsNoSuchEntityException() {
Assertions.assertThrows(
