SteNicholas commented on code in PR #3726:
URL: https://github.com/apache/celeborn/pull/3726#discussion_r3396339154
##########
common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala:
##########
@@ -6998,9 +6998,13 @@ object CelebornConf extends Logging {
buildConf("celeborn.client.shuffleDataLostOnUnknownWorker.enabled")
.categories("client")
.version("0.6.3")
- .doc("Whether to mark shuffle data lost when unknown worker is
detected.")
+ .doc("When enabled, any shuffle that had partitions on the (crashed) " +
+ "unknown worker is immediately marked as data lost. " +
+ "On the write flow revive/commit request for that shuffle will fast
fail. " +
+ "GetReducerFileGroup requests are replied with SHUFFLE_DATA_LOST. " +
+ "This has no effect when ${CLIENT_PUSH_REPLICATE_ENABLED.key}=true")
.booleanConf
- .createWithDefault(false)
+ .createWithDefault(true)
Review Comment:
**Default flip to `true` turns this on by default for the common case.**
`celeborn.client.push.replicate.enabled` defaults to `false`
(CelebornConf.scala:4856) and this feature is gated by `!pushReplicateEnabled`,
so flipping the default to `true` enables the aggressive fast-fail for every
non-replicated deployment on upgrade. Combined with the irreversibility noted
on `ReducePartitionCommitHandler.markShuffleDataLost`, a single master restart
can trigger cluster-wide unnecessary stage recomputes.
Granularity is also coarse: a shuffle is failed whenever the unknown worker
still appears as a key in `shuffleAllocatedWorkers`, even if its partitions
were already revived/migrated to healthy workers. Worth confirming the default
flip is intended for 0.7.0.
##########
common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala:
##########
@@ -6998,9 +6998,13 @@ object CelebornConf extends Logging {
buildConf("celeborn.client.shuffleDataLostOnUnknownWorker.enabled")
.categories("client")
.version("0.6.3")
- .doc("Whether to mark shuffle data lost when unknown worker is
detected.")
+ .doc("When enabled, any shuffle that had partitions on the (crashed) " +
+ "unknown worker is immediately marked as data lost. " +
+ "On the write flow revive/commit request for that shuffle will fast
fail. " +
+ "GetReducerFileGroup requests are replied with SHUFFLE_DATA_LOST. " +
+ "This has no effect when ${CLIENT_PUSH_REPLICATE_ENABLED.key}=true")
Review Comment:
This is a plain concatenated string (no `s` interpolator), so
`${CLIENT_PUSH_REPLICATE_ENABLED.key}` is emitted literally — it already shows
up verbatim in the generated `docs/configuration/client.md` row. Prefix the doc
string with `s` (as surrounding entries do) so it renders
`celeborn.client.push.replicate.enabled`.
##########
client/src/main/scala/org/apache/celeborn/client/commit/ReducePartitionCommitHandler.scala:
##########
@@ -144,22 +144,16 @@ class ReducePartitionCommitHandler(
if (mockShuffleLost) {
mockShuffleLostShuffle == shuffleId
} else {
- dataLostShuffleSet.contains(shuffleId) ||
isStageDataLostInUnknownWorker(shuffleId)
+ dataLostShuffleSet.contains(shuffleId)
}
}
- private def isStageDataLostInUnknownWorker(shuffleId: Int): Boolean = {
- if (conf.clientShuffleDataLostOnUnknownWorkerEnabled &&
!conf.clientPushReplicateEnabled) {
- val allocatedWorkers = shuffleAllocatedWorkers.get(shuffleId)
- if (allocatedWorkers != null) {
- return workerStatusTracker.excludedWorkers.asScala.collect {
- case (workerId, (status, _))
- if status == StatusCode.WORKER_UNKNOWN &&
allocatedWorkers.contains(workerId) =>
- workerId
- }.nonEmpty
- }
+ override def markShuffleDataLost(shuffleId: Int): Unit = {
+ logWarning(s"Marking shuffle $shuffleId data as lost due to
unknown/crashed worker.")
+ dataLostShuffleSet.add(shuffleId)
Review Comment:
**Irreversible marking vs. transient `WORKER_UNKNOWN`.**
`markShuffleDataLost` does `dataLostShuffleSet.add` + `setStageEnd`, and
neither is undone for the life of the shuffle (only `removeExpiredShuffle`
clears it). But `WORKER_UNKNOWN` is transient: the master computes
`unknownWorkers = needCheckedWorkerList.filterNot(workersMap.containsKey)`
(Master.scala:1241), so a still-alive worker that briefly leaves `workersMap` —
master failover/restart rebuilding state, a heartbeat-timeout eviction, or a
long GC pause — is reported unknown for a heartbeat and then recovers on
re-registration.
The removed `isStageDataLostInUnknownWorker` was evaluated live on every
`isStageDataLost` call and reverted once the worker left `excludedWorkers`.
With this change a transient blip **permanently** marks every affected shuffle
`SHUFFLE_DATA_LOST` and force-recomputes the stage, even though the committed
data is intact. Consider re-validating against current worker status before
failing, or keeping the mark reversible while the worker is only *unknown* (not
confirmed lost).
##########
client/src/main/scala/org/apache/celeborn/client/LifecycleManager.scala:
##########
@@ -86,6 +86,8 @@ class LifecycleManager(val appUniqueId: String, val conf:
CelebornConf) extends
private val pushRackAwareEnabled = conf.clientReserveSlotsRackAwareEnabled
private val partitionSplitThreshold = conf.shufflePartitionSplitThreshold
private val partitionSplitMode = conf.shufflePartitionSplitMode
+ private val shuffleDataLostOnUnknownWorkerEnabled =
Review Comment:
This field looks unused — `UnknownWorkerListener` keeps its own `private val
shuffleDataLostOnUnknownWorkerEnabled =
conf.clientShuffleDataLostOnUnknownWorkerEnabled` (in CommitManager) and reads
that. Either remove this one or route the listener through it.
##########
client/src/main/scala/org/apache/celeborn/client/CommitManager.scala:
##########
@@ -367,6 +371,31 @@ class CommitManager(appUniqueId: String, val conf:
CelebornConf, lifecycleManage
}
}
+ private class UnknownWorkerListener extends WorkerStatusListener {
+ private val shuffleDataLostOnUnknownWorkerEnabled =
+ conf.clientShuffleDataLostOnUnknownWorkerEnabled
+ private val pushReplicateEnabled = conf.clientPushReplicateEnabled
+
+ override def notifyChangedWorkersStatus(workersStatus: WorkersStatus):
Unit = {
+ if (shuffleDataLostOnUnknownWorkerEnabled && !pushReplicateEnabled) {
+ if (workersStatus.unknownWorkers != null &&
!workersStatus.unknownWorkers.isEmpty) {
+ lifecycleManager.shuffleAllocatedWorkers.asScala.foreach {
Review Comment:
This `foreach` has no per-shuffle guard. `markShuffleDataLost` →
`getCommitHandler(shuffleId)` → `lifecycleManager.getPartitionType(shuffleId)`;
if that is transiently null for a shuffle being torn down (or returns an
unexpected type → the `case _ => throw UnsupportedOperationException` in
`getCommitHandler`), the exception propagates out of
`notifyChangedWorkersStatus` and is only caught at
`WorkerStatusTracker.scala:207`, abandoning the rest of the loop so other
affected shuffles aren't marked this heartbeat. Since marking is per-shuffle,
wrap each iteration in its own try/catch (log and continue).
##########
docs/migration.md:
##########
@@ -37,6 +37,8 @@ license: |
- Since 0.7.0, Celeborn changed the default value of
`celeborn.port.maxRetries` from `1` to `16`.
+- Since 0.7.0, Celeborn change the default value of
`celeborn.client.shuffleDataLostOnUnknownWorker.enabled` from `false` to
`true`, which means Celeborn will treat shuffle data lost when unknown worker
is detected at default.
Review Comment:
Grammar: "Celeborn change" → "changed", and "treat shuffle data lost" →
"treat shuffle data as lost". Consider also adding the replication caveat (no
effect when `celeborn.client.push.replicate.enabled=true`) to match the config
doc.
--
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]
