This is an automated email from the ASF dual-hosted git repository.
SteNicholas 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 a647fe7ffb [CELEBORN-2494][TEST] Fix flaky PushDataTimeoutTest after
replica timeout
a647fe7ffb is described below
commit a647fe7ffb6af0cb3cf7bcaac88a7c233070398c
Author: zhengtao <[email protected]>
AuthorDate: Thu Oct 22 20:18:33 2026 +0800
[CELEBORN-2494][TEST] Fix flaky PushDataTimeoutTest after replica timeout
### What changes were proposed in this pull request?
JIRA: [CELEBORN-2494](https://issues.apache.org/jira/browse/CELEBORN-2494).
Enable `celeborn.client.excludePeerWorkerOnFailure.enabled` in the
`pushdata` and `pushMergeData` recovery cases in `PushDataTimeoutTest`.
### Why are the changes needed?
Since #3842, a replica push timeout adds the replica to the primary
worker's `unavailablePeers`. This suite keeps peers unavailable for 10 minutes
to assert that the timeout was recorded, but the two recovery cases explicitly
disable client-side peer exclusion. Revive can therefore randomly select the
same quarantined primary/replica pair again, exhaust the retry budget, and fail
the test.
In [this Spark 3.2.4 / JDK 8
failure](https://github.com/apache/celeborn/actions/runs/39197769991/job/117901785894),
partition 35 hit a replica timeout at 05:53:31.533 UTC. All five revive
attempts (epochs 1–5) selected the same primary/replica pair, which immediately
returned `PUSH_DATA_CREATE_CONNECTION_FAIL_REPLICA`; the task was aborted at
05:53:47.392 UTC.
Enabling peer exclusion keeps failed replicas out of revive candidates,
consistent with the production default and the existing excluded-worker test.
The injected primary/replica timeouts, retry budget, quarantine duration,
result comparisons, and `unavailablePeers` assertions are preserved.
### Does this PR resolve a correctness bug?
- [ ] Yes
This fixes flaky test configuration only.
### Does this PR introduce _any_ user-facing change?
- [ ] Yes
### How was this patch tested?
- `dev/reformat`: all 16 profile invocations succeeded; the changed Scala
test was covered.
- Spark 3.2.4 / Scala 2.12.18 / JDK 17.0.19: all five `PushDataTimeoutTest`
cases passed in three consecutive runs.
- Spark 3.5.8 / Scala 2.12.18 / JDK 17.0.19: all five cases passed.
- 20 successful test executions in total. The original JDK 8 CI environment
was not rerun locally.
Initial build and test for each Spark profile:
```sh
build/mvn -B -ntp -Pspark-3.2 -pl tests/spark-it -am install \
-Dtest=none
-DwildcardSuites=org.apache.celeborn.tests.spark.PushDataTimeoutTest
build/mvn -B -ntp -Pspark-3.5 -pl tests/spark-it -am install \
-Dtest=none
-DwildcardSuites=org.apache.celeborn.tests.spark.PushDataTimeoutTest
```
The two additional Spark 3.2.4 runs used `build/mvn -B -ntp -Pspark-3.2 -pl
tests/spark-it test` with the same test selectors.
Closes #3886 from zaynt4606/codex/fix-pushdata-timeout-flake.
Authored-by: zhengtao <[email protected]>
Signed-off-by: Nicholas Jiang <[email protected]>
---
.../scala/org/apache/celeborn/tests/spark/PushDataTimeoutTest.scala | 4 ++--
1 file changed, 2 insertions(+), 2 deletions(-)
diff --git
a/tests/spark-it/src/test/scala/org/apache/celeborn/tests/spark/PushDataTimeoutTest.scala
b/tests/spark-it/src/test/scala/org/apache/celeborn/tests/spark/PushDataTimeoutTest.scala
index 0163343073..fe5115a6e0 100644
---
a/tests/spark-it/src/test/scala/org/apache/celeborn/tests/spark/PushDataTimeoutTest.scala
+++
b/tests/spark-it/src/test/scala/org/apache/celeborn/tests/spark/PushDataTimeoutTest.scala
@@ -67,7 +67,7 @@ class PushDataTimeoutTest extends AnyFunSuite
.set(s"spark.${CelebornConf.CLIENT_PUSH_DATA_TIMEOUT.key}", "5s")
.set(s"spark.celeborn.data.push.timeoutCheck.interval", "2s")
.set(s"spark.${CelebornConf.CLIENT_PUSH_REPLICATE_ENABLED.key}",
enabled.toString)
-
.set(s"spark.${CelebornConf.CLIENT_EXCLUDE_PEER_WORKER_ON_FAILURE_ENABLED.key}",
"false")
+
.set(s"spark.${CelebornConf.CLIENT_EXCLUDE_PEER_WORKER_ON_FAILURE_ENABLED.key}",
"true")
// make sure PushDataHandler.handlePushData be triggered
.set(s"spark.${CelebornConf.CLIENT_PUSH_BUFFER_MAX_SIZE.key}", "5")
@@ -100,7 +100,7 @@ class PushDataTimeoutTest extends AnyFunSuite
.set(s"spark.${CelebornConf.CLIENT_PUSH_DATA_TIMEOUT.key}", "5s")
.set(s"spark.celeborn.data.push.timeoutCheck.interval", "2s")
.set(s"spark.${CelebornConf.CLIENT_PUSH_REPLICATE_ENABLED.key}",
enabled.toString)
-
.set(s"spark.${CelebornConf.CLIENT_EXCLUDE_PEER_WORKER_ON_FAILURE_ENABLED.key}",
"false")
+
.set(s"spark.${CelebornConf.CLIENT_EXCLUDE_PEER_WORKER_ON_FAILURE_ENABLED.key}",
"true")
val sparkSession = SparkSession.builder().config(sparkConf).getOrCreate()
val sqlResult = runsql(sparkSession)
