shlomitubul commented on code in PR #3721:
URL: https://github.com/apache/celeborn/pull/3721#discussion_r3394941805
##########
worker/src/main/scala/org/apache/celeborn/service/deploy/worker/Controller.scala:
##########
@@ -898,3 +905,54 @@ private[deploy] class Controller(
}
}
}
+
+private[deploy] object Controller {
+
+ def buildCommitFilesResponseOnCancel(
+ primaryIds: jList[String],
+ replicaIds: jList[String],
+ committedPrimaryIds: jSet[String],
+ committedReplicaIds: jSet[String],
+ emptyFilePrimaryIds: jSet[String],
+ emptyFileReplicaIds: jSet[String],
+ committedPrimaryStorageInfos: java.util.Map[String, StorageInfo],
+ committedReplicaStorageInfos: java.util.Map[String, StorageInfo],
+ committedMapIdBitMap: java.util.Map[String, RoaringBitmap],
+ partitionSizeList: java.util.Collection[Long]): CommitFilesResponse = {
+ // Commit tasks may still be running here: future.cancel(true) does not
interrupt a
+ // CompletableFuture. Compute failed = requested - committed - empty,
reading committed
+ // before snapshotting it below. The sets are append-only, so a partition
that commits in
+ // this window lands in both failed and committed (safe over-report),
never in neither --
+ // a partition in neither is read as empty-and-valid by the driver and
silently dropped.
+ val failedPrimaryIds = new jArrayList[String](primaryIds)
+ failedPrimaryIds.removeAll(committedPrimaryIds)
+ failedPrimaryIds.removeAll(emptyFilePrimaryIds)
+ val failedReplicaIds = new jArrayList[String](replicaIds)
+ failedReplicaIds.removeAll(committedReplicaIds)
+ failedReplicaIds.removeAll(emptyFileReplicaIds)
+ // COMMIT_FILE_EXCEPTION only when nothing committed and nothing empty;
empty files are a
+ // successful terminal state and must not be reported as failed.
+ val committedPrimaryIdList = new jArrayList[String](committedPrimaryIds)
+ val committedReplicaIdList = new jArrayList[String](committedReplicaIds)
+ // COMMIT_FILE_EXCEPTION only when nothing committed and nothing empty;
empty files are a
+ // successful terminal state and must not be reported as failed.
+ val status =
+ if (committedPrimaryIdList.isEmpty && committedReplicaIdList.isEmpty &&
+ emptyFilePrimaryIds.isEmpty && emptyFileReplicaIds.isEmpty) {
+ StatusCode.COMMIT_FILE_EXCEPTION
+ } else {
+ StatusCode.PARTIAL_SUCCESS
+ }
Review Comment:
Fixed in `9c3e848`. The status is now derived from the failed lists
materialized in the response: `SUCCESS` when both are empty,
`COMMIT_FILE_EXCEPTION` when nothing committed and nothing empty,
`PARTIAL_SUCCESS` otherwise. Verified the consequence you described:
`CommitHandler.doParallelCommitFiles` records the worker in
`commitFilesFailedWorkers` for any terminal status other than `SUCCESS`, which
flows into `WorkerStatusTracker.excludedWorkers` — so the empty-failed case did
penalize the worker with no actual failure. Returning `SUCCESS` there is safe
because the committed/empty sets are append-only: empty failed lists mean every
requested partition had already reached a terminal good state at snapshot time,
the same condition the normal `reply()` path uses. Added two unit tests
(all-empty, and cancellation racing with completion).
##########
worker/src/main/scala/org/apache/celeborn/service/deploy/worker/Controller.scala:
##########
@@ -705,18 +705,25 @@ private[deploy] class Controller(
case throwable: Throwable =>
logError(s"$errMsg, an unexpected exception occurred.",
throwable)
}
+ val response = Controller.buildCommitFilesResponseOnCancel(
+ primaryIds,
+ replicaIds,
+ committedPrimaryIds,
+ committedReplicaIds,
+ emptyFilePrimaryIds,
+ emptyFileReplicaIds,
+ committedPrimaryStorageInfos,
+ committedReplicaStorageInfos,
+ committedMapIdBitMap,
+ partitionSizeList)
commitInfo.synchronized {
- commitInfo.response = CommitFilesResponse(
- StatusCode.COMMIT_FILE_EXCEPTION,
- List.empty.asJava,
- List.empty.asJava,
- primaryIds,
- replicaIds)
-
+ commitInfo.response = response
commitInfo.status = CommitInfo.COMMIT_FINISHED
}
+ context.reply(response)
Review Comment:
Done in `9c3e848` — the cancel branch now releases slots and removes
partition locations before replying, mirroring `reply()`. Two corrections to
the premise, for the record: (1) the cleanup was not permanently skipped before
— `Worker.cleanup` releases the slots and removes the shuffle when the master
reports the shuffle key expired, independent of `commitInfo.status`, so the
window was "until shuffle unregistration", not forever; (2) this was
pre-existing behavior on the timeout path, not introduced here — before this PR
the branch also set `COMMIT_FINISHED` without cleanup, and the driver's
same-epoch re-ask hit the `COMMIT_FINISHED` short-circuit which replies the
cached response without releasing either. Still worth fixing for parity with
`reply()`, so it's in. One caveat documented in the code: unlike `reply()`,
commit tasks may still be running here (`cancel(true)` does not interrupt
them); a task that fetches its location after removal logs an error and marks
the par
tition failed, which is harmless since the response already reports every
not-committed-and-not-empty partition as failed, and the expiry-time release is
idempotent.
--
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]
