Copilot commented on code in PR #3721:
URL: https://github.com/apache/celeborn/pull/3721#discussion_r3428225068
##########
worker/src/main/scala/org/apache/celeborn/service/deploy/worker/Controller.scala:
##########
@@ -705,18 +709,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 =
+ partitionLocationInfo.removePrimaryPartitions(shuffleKey,
primaryIds)
+ val releaseReplicaLocations =
+ partitionLocationInfo.removeReplicaPartitions(shuffleKey,
replicaIds)
+ workerInfo.releaseSlots(shuffleKey, releasePrimaryLocations._1)
+ workerInfo.releaseSlots(shuffleKey, releaseReplicaLocations._1)
+ 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)
workerSource.incCounter(WorkerSource.COMMIT_FILES_FAIL_COUNT)
+ workerSource.stopTimer(WorkerSource.COMMIT_FILES_TIME,
shuffleKey)
Review Comment:
In the exceptional/cancel path, `COMMIT_FILES_FAIL_COUNT` is incremented
unconditionally, but `buildCommitFilesResponseOnCancel` can legitimately return
`StatusCode.SUCCESS` when cancellation races with completion (failed lists
empty). That will skew failure metrics/alerts even though the worker replied
success. Consider incrementing the fail counter only when the computed response
status is non-SUCCESS.
--
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]
