shlomitubul commented on code in PR #3721:
URL: https://github.com/apache/celeborn/pull/3721#discussion_r3409151523
##########
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:
Fixed in `a443eee`. Confirmed the scenario: the commit task body only
`catch`es `IOException`, so a non-`IOException` thrown by a task completes the
future exceptionally *without* the timer firing, and the `TimerTask` was never
cancelled on the error path → it fired later with the misleading `After waiting
… cancel all commit file jobs` warning (the cancels themselves were no-ops on
the already-completed future/tasks, so no correctness impact). Now
`timeout.cancel()` runs at the top of the error branch too, mirroring the
success path.
##########
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:
Done in `a443eee` — `totalWritten`/`fileCount` are now derived from a single
`partitionSizeList.asScala.toVector` snapshot. One correction for the record:
`size()` on a `LinkedBlockingQueue` is O(1) (an atomic counter), so this wasn't
two traversals — it was one traversal plus an O(1) read. And the practical
impact was negligible: both feed approximate size-estimation metrics
(`fileCount` already counts only partitions ≥ `minPartitionSizeToEstimate`), so
a concurrent `add` between the two reads could at most make them momentarily
off-by-one, never a correctness issue. Snapshotting once is a cheap consistency
tidy-up, so it's in.
--
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]
