Copilot commented on code in PR #3721:
URL: https://github.com/apache/celeborn/pull/3721#discussion_r3388309258
##########
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:
In the cancel/timeout branch, the code replies and marks the commit as
finished but never releases the reserved slots / removes the partition
locations (the normal `reply()` path does this via
`partitionLocationInfo.remove*Partitions` + `workerInfo.releaseSlots`). Since
`commitInfo.status` is set to `COMMIT_FINISHED`, this cleanup will never happen
later, which can leak `activeSlots` on the worker and impact subsequent slot
reservations.
##########
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:
`buildCommitFilesResponseOnCancel` can return `StatusCode.PARTIAL_SUCCESS`
even when `failedPrimaryIds`/`failedReplicaIds` are empty (e.g., all requested
partitions ended up committed or empty). That makes the response internally
inconsistent and can cause the client to record the worker in
`commitFilesFailedWorkers` despite no actual failures. Consider returning
`SUCCESS` when both failed lists are empty, and only using `PARTIAL_SUCCESS`
when there are real failures (keeping `COMMIT_FILE_EXCEPTION` for the ‘nothing
committed and nothing empty’ case).
--
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]
