This is an automated email from the ASF dual-hosted git repository.
HyukjinKwon pushed a commit to branch branch-4.x
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.x by this push:
new bc09f79d1fc4 [SPARK-57991][CORE][TEST] Deflake BarrierTaskContextSuite
wall-clock skew assertions on macOS-26
bc09f79d1fc4 is described below
commit bc09f79d1fc4dcbe48063eaa48edb2e47ca526ee
Author: Hyukjin Kwon <[email protected]>
AuthorDate: Tue Jul 7 16:46:40 2026 +0900
[SPARK-57991][CORE][TEST] Deflake BarrierTaskContextSuite wall-clock skew
assertions on macOS-26
### What changes were proposed in this pull request?
Widen the post-sync wall-clock skew tolerance in `BarrierTaskContextSuite`
from a hard-coded
`<= 1000` (ms) to a documented `maxSyncSkewMs = 3000` constant, applied to
the three tests that
assert how closely tasks finish a `barrier()` / `allGather()` global sync:
- `global sync by barrier() call`
- `successively sync with allGather and barrier`
- `support multiple barrier() call within a single task`
### Why are the changes needed?
The `Build / Maven (Scala 2.13, JDK 21, MacOS-26)` scheduled workflow has
been failing on every
recent run. One of the reproducing failures is:
```
BarrierTaskContextSuite:
- successively sync with allGather and barrier *** FAILED ***
1078 was not less than or equal to 1000
(BarrierTaskContextSuite.scala:122)
```
These tests capture `System.currentTimeMillis()` in each task immediately
after a
barrier/allGather returns and assert `times.max - times.min <= 1000`.
**That skew is bounded below
by the barrier client's own polling granularity, not by sync correctness.**
`BarrierTaskContext.runBarrier` waits for the coordinator RPC in a
`Thread.sleep(1000)` loop
([`BarrierTaskContext.scala:102`](https://github.com/apache/spark/blob/master/core/src/main/scala/org/apache/spark/BarrierTaskContext.scala#L102)),
so two tasks can observe the release up to ~1s apart from the poll interval
alone — before any
thread-scheduling or GC jitter on a busy CI host is added.
Evidence gathered:
- SPARK-49983 (2024, PR #48487) previously saw `1038` and only *halved the
pre-barrier sleep*.
That reduces arrival spread but cannot reduce poll-granularity skew, so
the flake persists on
slower runners like `macos-26` arm64.
- Reproduced locally on **macOS 26 arm64 / JDK 21** (the same OS/arch as
the failing runner):
measured post-sync skew up to **994 ms while the machine was idle** —
i.e. essentially zero
margin under the old `1000` bound, so any CI load tips it over.
The new bound (twice the 1 s poll interval plus a jitter allowance)
preserves the test's intent —
tasks stay loosely in lockstep and none can race an entire extra sync ahead
— without coupling the
assertion to a razor-thin wall-clock margin.
### Scope note
This PR fixes only the `BarrierTaskContextSuite` flake. The same workflow
also exhibits other,
independent flakes (Kafka `stress test for failOnDataLoss=false`
topic-deletion timeout, mllib GMM
FP-tolerance on arm64, `StateStoreSuite.maintenance` timeout). Those are
out of scope here and
would be separate changes. (The earlier `AF_UNIX path too long` UDF-worker
flake was already fixed
upstream by SPARK-57949.)
### Does this PR introduce _any_ user-facing change?
No. Test-only.
### How was this patch tested?
Locally on macOS 26 arm64 / JDK 21:
`build/mvn -pl core test
-DwildcardSuites=org.apache.spark.scheduler.BarrierTaskContextSuite`
— all 13 tests pass; scalastyle clean.
Also validated end-to-end on the `Build / Maven (Scala 2.13, JDK 21,
MacOS-26)` workflow,
running the same `core` matrix job on the identical `macos-26` arm64 runner
image:
| | `core,launcher,…` job (runs `BarrierTaskContextSuite`) | Result |
|---|---|---|
| **Before** (`apache/spark` master, without this fix) | [runs/28790673283
· job
85379029552](https://github.com/apache/spark/actions/runs/28790673283/job/85379029552)
| ❌ `1078 was not less than or equal to 1000
(BarrierTaskContextSuite.scala:122)` — `Tests: succeeded 4103, failed 1` |
| **After** (this branch) | [runs/28829214565 · job
85505367410](https://github.com/HyukjinKwon/spark/actions/runs/28829214565/job/85505367410)
| ✅ `Tests: succeeded 4104, failed 0`; the three barrier tests all pass |
(To run the guarded workflow on the fork, its `if: github.repository ==
'apache/spark'`
condition was relaxed on a throwaway commit that is **not** part of this PR
— this PR is the
single-file test change only.)
### Was this patch authored or co-authored using generative AI tooling?
Yes, drafted with assistance from Claude.
Closes #57057 from HyukjinKwon/deflake-barrier-macos26.
Authored-by: Hyukjin Kwon <[email protected]>
Signed-off-by: Hyukjin Kwon <[email protected]>
(cherry picked from commit e887fd5680977c72b42ba83dc605c551aceb8ed0)
Signed-off-by: Hyukjin Kwon <[email protected]>
---
.../spark/scheduler/BarrierTaskContextSuite.scala | 22 +++++++++++++++++-----
1 file changed, 17 insertions(+), 5 deletions(-)
diff --git
a/core/src/test/scala/org/apache/spark/scheduler/BarrierTaskContextSuite.scala
b/core/src/test/scala/org/apache/spark/scheduler/BarrierTaskContextSuite.scala
index f00fb0d2cfa3..ee96bf614809 100644
---
a/core/src/test/scala/org/apache/spark/scheduler/BarrierTaskContextSuite.scala
+++
b/core/src/test/scala/org/apache/spark/scheduler/BarrierTaskContextSuite.scala
@@ -31,6 +31,18 @@ import
org.apache.spark.internal.config.Tests.TEST_NO_STAGE_RETRY
class BarrierTaskContextSuite extends SparkFunSuite with LocalSparkContext
with Eventually {
+ // Upper bound (in millis) on the wall-clock skew we tolerate between tasks
right after a
+ // `barrier()` / `allGather()` global sync returns. The sync releases all
tasks together, but
+ // each task only observes the release the next time
`BarrierTaskContext.runBarrier` wakes from
+ // its `Thread.sleep(1000)` polling loop, so two tasks can record
`System.currentTimeMillis()`
+ // up to ~1s apart from the poll granularity alone -- before adding
thread-scheduling and GC
+ // jitter on a busy CI host. The historical `<= 1000` bound left no room for
that and flaked on
+ // slower runners (e.g. macOS 26 arm64 observed 1078ms; SPARK-49983 earlier
observed 1038ms).
+ // Use twice the poll interval plus a jitter allowance; the test still
asserts the two rounds
+ // stay loosely in lockstep (a task cannot race a whole extra sync ahead)
without coupling to a
+ // razor-thin wall-clock margin.
+ private val maxSyncSkewMs = 3000
+
def initLocalClusterSparkContext(numWorker: Int = 4, conf: SparkConf = new
SparkConf()): Unit = {
conf
// Init local cluster here so each barrier task runs in a separated
process, thus `barrier()`
@@ -55,7 +67,7 @@ class BarrierTaskContextSuite extends SparkFunSuite with
LocalSparkContext with
val times = rdd2.collect()
// All the tasks shall finish global sync within a short time slot.
- assert(times.max - times.min <= 1000)
+ assert(times.max - times.min <= maxSyncSkewMs)
}
test("share messages with allGather() call") {
@@ -115,11 +127,11 @@ class BarrierTaskContextSuite extends SparkFunSuite with
LocalSparkContext with
val times = rdd2.collect()
// All the tasks shall finish the first round of global sync within a
short time slot.
val times1 = times.map(_._1)
- assert(times1.max - times1.min <= 1000)
+ assert(times1.max - times1.min <= maxSyncSkewMs)
// All the tasks shall finish the second round of global sync within a
short time slot.
val times2 = times.map(_._2)
- assert(times2.max - times2.min <= 1000)
+ assert(times2.max - times2.min <= maxSyncSkewMs)
}
test("support multiple barrier() call within a single task") {
@@ -141,11 +153,11 @@ class BarrierTaskContextSuite extends SparkFunSuite with
LocalSparkContext with
val times = rdd2.collect()
// All the tasks shall finish the first round of global sync within a
short time slot.
val times1 = times.map(_._1)
- assert(times1.max - times1.min <= 1000)
+ assert(times1.max - times1.min <= maxSyncSkewMs)
// All the tasks shall finish the second round of global sync within a
short time slot.
val times2 = times.map(_._2)
- assert(times2.max - times2.min <= 1000)
+ assert(times2.max - times2.min <= maxSyncSkewMs)
}
test("throw exception on barrier() call timeout") {
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
