This is an automated email from the ASF dual-hosted git repository.
zaynt4606 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/celeborn.git
The following commit(s) were added to refs/heads/main by this push:
new 00113daf3a [CELEBORN-2401] Change default of
celeborn.client.spark.fetch.cleanFailedShuffle to true
00113daf3a is described below
commit 00113daf3a5d2476aaa04a8b576152893d377fda
Author: wang-haihua <[email protected]>
AuthorDate: Mon Aug 10 17:24:35 2026 +0800
[CELEBORN-2401] Change default of
celeborn.client.spark.fetch.cleanFailedShuffle to true
### What changes were proposed in this pull request?
Flip the default of `celeborn.client.spark.fetch.cleanFailedShuffle` from
`false` to `true`.
Only two lines change:
- `common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala` —
`createWithDefault(false)` → `createWithDefault(true)` on
`CLIENT_FETCH_CLEAN_FAILED_SHUFFLE`.
- `docs/configuration/client.md` — the golden config table is updated in
the same commit so `ConfigurationSuite` stays green.
### Why are the changes needed?
`celeborn.client.spark.fetch.cleanFailedShuffle` was introduced in 0.6.0 by
[CELEBORN-1896](https://issues.apache.org/jira/browse/CELEBORN-1896) (PR #3109)
and hardened in 0.7.0 by
[CELEBORN-2088](https://issues.apache.org/jira/browse/CELEBORN-2088) (PR
#3401). When enabled:
1. The driver records shuffleIds that hit `FetchFailed`.
2. When the retried stage registers a new shuffleId, the previously failed
shuffleId is enqueued for cleanup by `FailedShuffleCleaner`.
3. `FailedShuffleCleaner` reuses the existing `UnregisterShuffle` path to
reclaim the stale shuffle data on workers.
Since 0.4.0 `celeborn.client.spark.stageRerun.enabled` has defaulted to
`true`, so `FetchFailed` stage rerun is already the out-of-the-box behavior.
However `cleanFailedShuffle` still defaults to `false`, so every retried stage
silently leaves its old shuffle data on the workers until the app exits or the
periodic expired-shuffle cleanup catches it. The two defaults contradict each
other.
A recent production incident on our cluster:
- A single stage retried three times.
- Retry 1: wrote ~143 TiB of shuffle data before failing.
- Retry 2: wrote another ~143 TiB before failing.
- Retry 3: was still running, ~61 TiB written so far.
- **300+ TiB of stale shuffle bytes were sitting on the workers
simultaneously**, none of which the job would ever fetch again.
On a shared cluster this drives workers into disk-full / high-usage
exclusion, cascading into more `FetchFailed` events for unrelated applications
and destabilizing the cluster until the offending app exits. Flipping the
default removes this failure mode without adding any new deletion path.
The proposal was raised on dev and supported there; see the [discussion
thread](https://lists.apache.org/list.html?devceleborn.apache.org) for context.
### Does this PR introduce _any_ user-facing change?
Yes — behavior change with an explicit escape hatch.
- **Before**: after a `FetchFailed`-driven stage rerun, the failed
shuffleId's data lingered on the workers until the application exited (or
`celeborn.worker.shuffle.expired.timeout` caught it).
- **After**: the failed shuffleId's data is reclaimed as soon as the
retried stage registers, via the existing `UnregisterShuffle` path.
- **Rollback**: users who need the previous behavior can set
`celeborn.client.spark.fetch.cleanFailedShuffle=false` in their Spark
configuration.
The release notes should call out both the new default and the rollback
flag.
### How was this patch tested?
- `ConfigurationSuite` — golden `docs/configuration/client.md` regenerated
to keep the doc test green.
- `CelebornFetchFailureDiskCleanSuite` — the end-to-end integration test
for this path already sets
`spark.celeborn.client.spark.fetch.cleanFailedShuffle=true` explicitly and
continues to pass; it exercises the exact code path affected by this default
flip.
- No new deletion route or code path is introduced; the mechanism is the
existing `FailedShuffleCleaner` → `UnregisterShuffle` flow that has shipped
since 0.6.0.
Related JIRA:
[CELEBORN-2401](https://issues.apache.org/jira/browse/CELEBORN-2401).
Closes #3779 from wang-haihua/flip-cleanfailedshuffle-default.
Authored-by: wang-haihua <[email protected]>
Signed-off-by: zhengtao <[email protected]>
AI-Contributed/Feature: 0/6
AI-Contributed/UT: 0/2
---
common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala | 2 +-
docs/configuration/client.md | 2 +-
docs/migration.md | 2 ++
.../org/apache/celeborn/tests/spark/CelebornFetchFailureSuite.scala | 1 +
.../org/apache/celeborn/tests/spark/CelebornIntegrityCheckSuite.scala | 1 +
5 files changed, 6 insertions(+), 2 deletions(-)
diff --git
a/common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala
b/common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala
index ea6b819fcf..6dce12fec6 100644
--- a/common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala
+++ b/common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala
@@ -5258,7 +5258,7 @@ object CelebornConf extends Logging {
.version("0.6.0")
.doc("whether to clean those disk space occupied by shuffles which
cannot be fetched")
.booleanConf
- .createWithDefault(false)
+ .createWithDefault(true)
val CLIENT_FETCH_CLEAN_FAILED_SHUFFLE_INTERVAL: ConfigEntry[Long] =
buildConf("celeborn.client.spark.fetch.cleanFailedShuffleInterval")
diff --git a/docs/configuration/client.md b/docs/configuration/client.md
index 5d15c6859f..be5422c236 100644
--- a/docs/configuration/client.md
+++ b/docs/configuration/client.md
@@ -125,7 +125,7 @@ license: |
| celeborn.client.shuffleDataLostOnUnknownWorker.enabled | false | false |
Whether to mark shuffle data lost when unknown worker is detected. | 0.6.3 | |
| celeborn.client.slot.assign.maxWorkers | 10000 | false | Max workers that
slots of one shuffle can be allocated on. Will choose the smaller positive one
from Master side and Client side, see `celeborn.master.slot.assign.maxWorkers`.
| 0.3.1 | |
| celeborn.client.spark.batch.openStream.parallelClientCreation.enabled | true
| false | Whether to create data clients in parallel before sending Spark batch
open-stream requests. When false, data clients are created serially. | 0.6.3 |
|
-| celeborn.client.spark.fetch.cleanFailedShuffle | false | false | whether to
clean those disk space occupied by shuffles which cannot be fetched | 0.6.0 |
|
+| celeborn.client.spark.fetch.cleanFailedShuffle | true | false | whether to
clean those disk space occupied by shuffles which cannot be fetched | 0.6.0 |
|
| celeborn.client.spark.fetch.cleanFailedShuffleInterval | 1s | false | the
interval to clean the failed-to-fetch shuffle files, only valid when
celeborn.client.spark.fetch.cleanFailedShuffle is enabled | 0.6.0 | |
| celeborn.client.spark.push.dynamicWriteMode.enabled | false | false |
Whether to dynamically switch push write mode based on conditions.If true,
shuffle mode will be only determined by partition count | 0.5.0 | |
| celeborn.client.spark.push.dynamicWriteMode.partitionNum.threshold | 2000 |
false | Threshold of shuffle partition number for dynamically switching push
writer mode. When the shuffle partition number is greater than this value, use
the sort-based shuffle writer for memory efficiency; otherwise use the
hash-based shuffle writer for speed. This configuration only takes effect when
celeborn.client.spark.push.dynamicWriteMode.enabled is true. | 0.5.0 | |
diff --git a/docs/migration.md b/docs/migration.md
index 8b3625f446..75d6931935 100644
--- a/docs/migration.md
+++ b/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 changed the default value of
`celeborn.client.spark.fetch.cleanFailedShuffle` from `false` to `true`, which
means Celeborn will reclaim shuffle data of `FetchFailed` retried stages at
default.
+
# Upgrading from 0.5 to 0.6
- Since 0.6.0, Celeborn deprecate
`celeborn.client.spark.fetch.throwsFetchFailure`. Please use
`celeborn.client.spark.stageRerun.enabled` instead.
diff --git
a/tests/spark-it/src/test/scala/org/apache/celeborn/tests/spark/CelebornFetchFailureSuite.scala
b/tests/spark-it/src/test/scala/org/apache/celeborn/tests/spark/CelebornFetchFailureSuite.scala
index 49558dfe6f..1a390aa9ae 100644
---
a/tests/spark-it/src/test/scala/org/apache/celeborn/tests/spark/CelebornFetchFailureSuite.scala
+++
b/tests/spark-it/src/test/scala/org/apache/celeborn/tests/spark/CelebornFetchFailureSuite.scala
@@ -93,6 +93,7 @@ class CelebornFetchFailureSuite extends AnyFunSuite
.config("spark.sql.shuffle.partitions", 2)
.config("spark.celeborn.shuffle.forceFallback.partition.enabled",
false)
.config("spark.celeborn.client.spark.stageRerun.enabled", "false")
+ .config("spark.celeborn.client.spark.fetch.cleanFailedShuffle",
"false")
.getOrCreate()
val value = Range(1, 10000).mkString(",")
diff --git
a/tests/spark-it/src/test/scala/org/apache/celeborn/tests/spark/CelebornIntegrityCheckSuite.scala
b/tests/spark-it/src/test/scala/org/apache/celeborn/tests/spark/CelebornIntegrityCheckSuite.scala
index daea52ac1d..a10654e8b0 100644
---
a/tests/spark-it/src/test/scala/org/apache/celeborn/tests/spark/CelebornIntegrityCheckSuite.scala
+++
b/tests/spark-it/src/test/scala/org/apache/celeborn/tests/spark/CelebornIntegrityCheckSuite.scala
@@ -90,6 +90,7 @@ class CelebornIntegrityCheckSuite extends AnyFunSuite
s"spark.${CelebornConf.SHUFFLE_COMPRESSION_CODEC.key}",
CompressionCodec.NONE.toString)
.config(s"spark.${CelebornConf.CLIENT_STAGE_RERUN_ENABLED.key}",
"false")
+
.config(s"spark.${CelebornConf.CLIENT_FETCH_CLEAN_FAILED_SHUFFLE.key}", "false")
.config("spark.celeborn.client.shuffle.integrityCheck.enabled", "true")
.config(
"spark.shuffle.manager",
