Copilot commented on code in PR #3721:
URL: https://github.com/apache/celeborn/pull/3721#discussion_r3395843529
##########
worker/src/main/scala/org/apache/celeborn/service/deploy/worker/Controller.scala:
##########
@@ -898,3 +917,59 @@ 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)
+ // 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
+ }
+ CommitFilesResponse(
+ status,
+ committedPrimaryIdList,
+ committedReplicaIdList,
+ failedPrimaryIds,
+ failedReplicaIds,
+ new jHashMap[String, StorageInfo](committedPrimaryStorageInfos),
+ new jHashMap[String, StorageInfo](committedReplicaStorageInfos),
+ new jHashMap[String, RoaringBitmap](committedMapIdBitMap),
+ partitionSizeList.asScala.sum,
+ partitionSizeList.size())
Review Comment:
`buildCommitFilesResponseOnCancel` computes `totalWritten` and `fileCount`
via `partitionSizeList.asScala.sum` and `partitionSizeList.size()`, which can
traverse the collection twice and (because tasks may still be adding to the
queue) can yield inconsistent results between the two reads. Compute both in a
single pass to avoid extra work and keep the fields internally consistent.
##########
worker/src/main/scala/org/apache/celeborn/service/deploy/worker/Controller.scala:
##########
@@ -705,18 +705,37 @@ private[deploy] class Controller(
case throwable: Throwable =>
logError(s"$errMsg, an unexpected exception occurred.",
throwable)
}
+ // Release slots and remove partition locations before reply,
mirroring reply().
+ // Unlike reply(), commit tasks may still be running here
(cancel(true) does not
+ // interrupt them): a task that has not yet fetched its location
will then find it
+ // removed and mark the partition failed, which is harmless --
the response below
+ // already reports every not-committed-and-not-empty partition
as failed, and the
+ // shuffle-expiry cleanup that previously reclaimed these slots
is idempotent.
+ val releasePrimaryLocations =
Review Comment:
In the async error/cancel branch, the `timeout` TimerTask is only cancelled
on the success path (`t == null`). If the commit `future` completes
exceptionally for reasons other than the timer (e.g., an unexpected runtime
exception inside a commit task), the timeout task will still fire later and log
a misleading warning + attempt to cancel already-finished work. Cancel the
timeout once the future completes on the error path as well.
--
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]
