sunchao commented on code in PR #3721:
URL: https://github.com/apache/celeborn/pull/3721#discussion_r3444357430
##########
worker/src/main/scala/org/apache/celeborn/service/deploy/worker/Controller.scala:
##########
@@ -898,3 +921,62 @@ 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)
+ val committedPrimaryIdList = new jArrayList[String](committedPrimaryIds)
+ val committedReplicaIdList = new jArrayList[String](committedReplicaIds)
+ // Snapshot once: tasks may still be adding, so deriving totalWritten and
fileCount from a
+ // single traversal keeps the two fields mutually consistent.
+ val partitionSizes = partitionSizeList.asScala.toVector
+ // Derive status from the failed lists returned in this response. Empty
failed lists mean
+ // every requested partition reached a terminal good state before the
snapshot (committed
+ // or empty -- both sets are append-only), which is the normal path's
SUCCESS condition;
+ // replying PARTIAL_SUCCESS would needlessly land this worker in the
client's
+ // commitFilesFailedWorkers. Empty files are a successful terminal state
and must not be
+ // reported as failed, so COMMIT_FILE_EXCEPTION is returned only when
nothing committed
+ // and nothing is empty.
+ val status =
+ if (failedPrimaryIds.isEmpty && failedReplicaIds.isEmpty) {
+ StatusCode.SUCCESS
+ } else if (committedPrimaryIdList.isEmpty &&
committedReplicaIdList.isEmpty &&
+ emptyFilePrimaryIds.isEmpty && emptyFileReplicaIds.isEmpty) {
+ StatusCode.COMMIT_FILE_EXCEPTION
+ } else {
+ StatusCode.PARTIAL_SUCCESS
Review Comment:
[P1] Preserve timeout exclusions for PARTIAL_SUCCESS
Returning `PARTIAL_SUCCESS` here changes the worker-exclusion lifetime.
`CommitHandler` treats every non-`SUCCESS` terminal response as a failed
worker, but `WorkerStatusTracker.handleHeartbeatResponse` retains
`COMMIT_FILE_EXCEPTION` and `WORKER_UNRESPONSIVE` for
`celeborn.client.excludedWorker.expireTimeout` while `PARTIAL_SUCCESS` falls
through to removal on the next healthy application heartbeat (10s by default
versus the 180s exclusion window). `ChangePartitionManager` then admits the
same worker again through `workerAvailable`.
For a timeout with partial progress, this can reselect the same overloaded
worker for the recomputed stage—the CELEBORN-2009 retry loop that the previous
`COMMIT_FILE_EXCEPTION` status avoided. Please either retain `PARTIAL_SUCCESS`
in `WorkerStatusTracker` (with a heartbeat regression test and a client-first
rollout), or keep a retained failure status on this timeout path while still
carrying the accurate committed/failed lists.
--
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]
