This is an automated email from the ASF dual-hosted git repository.
yuqi1129 pushed a commit to branch branch-1.3
in repository https://gitbox.apache.org/repos/asf/gravitino.git
The following commit(s) were added to refs/heads/branch-1.3 by this push:
new 706cfbe68f [Cherry-pick to branch-1.3] [#12669] improvement(core): Use
entityStore.update() for job status transitions (#12675) (#12717)
706cfbe68f is described below
commit 706cfbe68f099d89457950289df9a8f31d119ff5
Author: github-actions[bot]
<41898282+github-actions[bot]@users.noreply.github.com>
AuthorDate: Sat Aug 29 10:00:03 2026 +0800
[Cherry-pick to branch-1.3] [#12669] improvement(core): Use
entityStore.update() for job status transitions (#12675) (#12717)
**Cherry-pick Information:**
- Original commit: 12ddb878948ddacea29f66754fb9d81ffa607ce9
- Target branch: `branch-1.3`
- Status: ✅ Conflicts resolved
**Conflict resolution notes:**
`branch-1.3` doesn't have the `startedAt` field on `JobEntity`/`JobPO`
(only `finishedAt` exists here), and predates the
`insertEntity`/`updateEntity`/`deleteEntity` helper refactor in
`JDBCBackend`, so the original diff couldn't apply cleanly. Resolved by
dropping the `startedAt`-specific logic and tests while keeping the
actual fix intact: `entityStore.update()` with CAS-based optimistic
concurrency for job status transitions, the `NoSuchEntityException` ->
`NoSuchJobException` translation in `cancelJob`, and the per-job
skip-and-warn handling in `pullAndUpdateJobStatus` so a concurrently
deleted job can't cancel all future status polls.
---------
Co-authored-by: Jerry Shao <[email protected]>
Co-authored-by: Claude Sonnet 5 <[email protected]>
Co-authored-by: Jerry Shao <[email protected]>
---
.../java/org/apache/gravitino/job/JobManager.java | 186 ++++++++++-----
.../gravitino/storage/relational/JDBCBackend.java | 2 +
.../storage/relational/mapper/JobMetaMapper.java | 3 +
.../mapper/JobMetaSQLProviderFactory.java | 5 +
.../provider/base/JobMetaBaseSQLProvider.java | 16 ++
.../gravitino/storage/relational/po/JobPO.java | 32 +++
.../storage/relational/service/JobMetaService.java | 75 ++++--
.../org/apache/gravitino/job/TestJobManager.java | 258 ++++++++++++++++++++-
.../gravitino/storage/relational/po/TestJobPO.java | 58 +++++
.../relational/service/TestJobMetaService.java | 140 +++++++++++
10 files changed, 706 insertions(+), 69 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 efa24d70ea..44d1bf2fde 100644
--- a/core/src/main/java/org/apache/gravitino/job/JobManager.java
+++ b/core/src/main/java/org/apache/gravitino/job/JobManager.java
@@ -501,37 +501,63 @@ public class JobManager implements JobOperationDispatcher
{
}
// Update the job status to CANCELING
- JobEntity newJobEntity =
- JobEntity.builder()
- .withId(jobEntity.id())
- .withJobExecutionId(jobEntity.jobExecutionId())
- .withJobTemplateName(jobEntity.jobTemplateName())
- .withStatus(JobHandle.Status.CANCELLING)
- .withNamespace(jobEntity.namespace())
- .withAuditInfo(
- AuditInfo.builder()
- .withCreator(jobEntity.auditInfo().creator())
- .withCreateTime(jobEntity.auditInfo().createTime())
-
.withLastModifier(PrincipalUtils.getCurrentPrincipal().getName())
- .withLastModifiedTime(Instant.now())
- .build())
- .build();
return TreeLockUtils.doWithTreeLock(
NameIdentifierUtil.ofJob(metalake, jobId),
LockType.WRITE,
() -> {
try {
- // 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
+ // 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 %s to CANCELING
status", newJobEntity),
+ String.format("Failed to update job entity for job %s to
CANCELING status", jobId),
e);
}
});
}
+ 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
finishedAt the job already
+ // had.
+ .withFinishedAt(latestJobEntity.finishedAt())
+ .build();
+ }
+
@Override
public void close() throws IOException {
try {
@@ -588,50 +614,106 @@ public class JobManager implements
JobOperationDispatcher {
}
if (newStatus != job.status()) {
- 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())
- .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
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;
+ }
+
+ // 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() != null &&
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())
+ .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 f401cd50be..65182cf57e 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
@@ -259,6 +259,8 @@ public class JDBCBackend implements RelationalBackend {
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 054c093311..030550311a 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
@@ -134,6 +134,22 @@ 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_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 8cb9d2e379..4c787f0685 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,38 @@ 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())
+ .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 5edbbc18ef..017bc6c21d 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;
@@ -83,6 +85,7 @@ import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
+import org.mockito.ArgumentCaptor;
import org.mockito.MockedStatic;
import org.mockito.Mockito;
@@ -560,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());
@@ -600,12 +603,91 @@ 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())
+ .withFinishedAt(67890L)
+ .build();
+ stubEntityStoreUpdateToApply(latestSucceeded);
+
+ JobEntity result = jobManager.cancelJob(metalake, queuedSnapshot.name());
+ Assertions.assertEquals(JobHandle.Status.SUCCEEDED, result.status());
+ 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())
+ .withFinishedAt(0L)
+ .build();
+ stubEntityStoreUpdateToApply(latestCancelling);
+
+ JobEntity result = jobManager.cancelJob(metalake, queuedSnapshot.name());
+ Assertions.assertEquals(JobHandle.Status.CANCELLING, result.status());
+ Assertions.assertEquals(0L, result.finishedAt());
+ }
+
@Test
public void testPullJobStatus() throws IOException {
JobEntity job = newJobEntity("shell_job", JobHandle.Status.QUEUED);
@@ -628,11 +710,157 @@ 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());
- verify(entityStore, times(1)).put(any(JobEntity.class), anyBoolean());
+
+ // Once a job transitions to a terminal status, finishedAt must be set.
+ JobEntity updatedJob = captureUpdatedJobEntity(job);
+ Assertions.assertEquals(JobHandle.Status.SUCCEEDED, updatedJob.status());
+ Assertions.assertNotNull(updatedJob.finishedAt());
+ Assertions.assertTrue(updatedJob.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
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())
+ .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(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.
+ 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())
+ .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(0L, result.finishedAt());
}
@Test
@@ -942,6 +1170,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 3fd8f9988b..d105fab5e1 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
@@ -141,4 +141,62 @@ public class TestJobPO {
Assertions.assertEquals(jobEntity.namespace(), resultEntity.namespace());
Assertions.assertEquals(jobEntity.auditInfo().creator(),
resultEntity.auditInfo().creator());
}
+
+ @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())
+ .withFinishedAt(0L)
+ .build();
+
+ JobPO.JobPOBuilder initBuilder = JobPO.builder().withMetalakeId(1L);
+ JobPO oldJobPO = JobPO.initializeJobPO(oldEntity, initBuilder);
+
+ long finishedAt = 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.SUCCEEDED)
+ .withNamespace(oldEntity.namespace())
+ .withAuditInfo(updatedAuditInfo)
+ .withFinishedAt(finishedAt)
+ .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.SUCCEEDED.name(),
newJobPO.jobRunStatus());
+ Assertions.assertEquals(finishedAt, 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.SUCCEEDED, resultEntity.status());
+ Assertions.assertEquals(finishedAt, resultEntity.finishedAt());
+ 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 9136d5561f..b073a4c6f4 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
@@ -227,6 +227,146 @@ 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 a new audit info.
+ 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)
+ .withFinishedAt(oldJob.finishedAt())
+ .build());
+
+ Assertions.assertEquals(JobHandle.Status.STARTED, startedJob.status());
+ 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 = System.currentTimeMillis();
+ 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())
+ .withFinishedAt(finishedAt)
+ .build());
+
+ Assertions.assertEquals(JobHandle.Status.SUCCEEDED, finishedJob.status());
+ 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())
+ .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(
