jerryshao commented on code in PR #12675:
URL: https://github.com/apache/gravitino/pull/12675#discussion_r3871634249
##########
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:
I see, let me fix it.
--
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]
