Copilot commented on code in PR #3721:
URL: https://github.com/apache/celeborn/pull/3721#discussion_r3375839916
##########
worker/src/main/scala/org/apache/celeborn/service/deploy/worker/Controller.scala:
##########
@@ -898,3 +905,50 @@ 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 status =
+ if (committedPrimaryIds.isEmpty && committedReplicaIds.isEmpty &&
+ emptyFilePrimaryIds.isEmpty && emptyFileReplicaIds.isEmpty) {
+ StatusCode.COMMIT_FILE_EXCEPTION
+ } else {
+ StatusCode.PARTIAL_SUCCESS
+ }
+ CommitFilesResponse(
+ status,
+ new jArrayList[String](committedPrimaryIds),
+ new jArrayList[String](committedReplicaIds),
+ failedPrimaryIds,
+ failedReplicaIds,
+ new jHashMap[String, StorageInfo](committedPrimaryStorageInfos),
+ new jHashMap[String, StorageInfo](committedReplicaStorageInfos),
+ new jHashMap[String, RoaringBitmap](committedMapIdBitMap),
+ partitionSizeList.asScala.sum,
+ partitionSizeList.size())
Review Comment:
`status` is computed from the live `committed*Ids` sets, but the response’s
`committed*Ids` lists are materialized afterwards. If a partition commits
between the `status` check and `new jArrayList(committed*Ids)`, the response
can become internally inconsistent (e.g., `COMMIT_FILE_EXCEPTION` while
returning non-empty committed lists), which can affect driver-side
handling/metrics. Compute `status` from the same snapshots you return to
guarantee consistency.
--
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]
