yuqi1129 commented on code in PR #12675:
URL: https://github.com/apache/gravitino/pull/12675#discussion_r3871476761
##########
core/src/main/java/org/apache/gravitino/job/JobManager.java:
##########
@@ -625,41 +633,46 @@ void pullAndUpdateJobStatus() {
? 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 identity/audit
+ // fields it carries forward can't clobber a concurrent change
made in the gap
+ // between the listJobs() snapshot above and this point (e.g. by
cancelJob()).
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);
- }
- });
+ boolean finalIsFinished = isFinished;
+ try {
+ 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,
startedAt, finalIsFinished));
Review Comment:
The updater receives the latest entity, but `finalNewStatus`, `startedAt`,
and `finalIsFinished` were computed from the earlier `listJobs()` snapshot. In
a multi-node race, another writer may already have moved the row to
`CANCELLING` or a terminal state, or recorded an earlier
`startedAt`/`finishedAt`, before this update reads it. This lambda can then
regress the latest status or overwrite those timestamps even though the SQL CAS
itself is correct. Could we derive and validate the transition inside the
updater from `latestJobEntity`, preserving terminal/cancelling states and
already-recorded timestamps?
##########
core/src/test/java/org/apache/gravitino/job/TestJobManager.java:
##########
@@ -782,19 +799,69 @@ public void
testPullJobStatusStartedAtNotBackfilledOnDirectCancellation() throws
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 {
Review Comment:
Could we add a race-focused test where `listJobs()` returns a stale
`QUEUED`/`STARTED` snapshot while `EntityStore.update()` invokes the updater
with a different, newer entity? At minimum, please verify that a latest
`CANCELLING` or terminal state is not regressed and that an already-recorded
`startedAt`/`finishedAt` is preserved. The current helpers pass the same entity
as both the list snapshot and the latest entity, so they cannot catch the
stale-merge issue above.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
