This is an automated email from the ASF dual-hosted git repository.
HyukjinKwon pushed a commit to branch branch-4.2
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.2 by this push:
new eb88a2bac747 [SPARK-57959][SQL][TEST] Deflake
MetricsFailureInjectionSuite non-deterministic injection test
eb88a2bac747 is described below
commit eb88a2bac7479e91b2048403f3411027bc031442
Author: Hyukjin Kwon <[email protected]>
AuthorDate: Mon Jul 6 18:55:02 2026 +0900
[SPARK-57959][SQL][TEST] Deflake MetricsFailureInjectionSuite
non-deterministic injection test
### What changes were proposed in this pull request?
Make `MetricsFailureInjectionSuite`'s `Non-deterministic stage block
failure injection -
injectFailure=true` case re-run its query (resetting the metrics each
attempt) until the
injected shuffle fetch-failure actually forces a stage-1 recompute, up to
10 attempts, before
running the existing assertions. Test-only.
### Why are the changes needed?
The test is flaky. In the full suite it fails ~1 run in 6 with:
```
- Non-deterministic stage block failure injection - injectFailure=true ***
FAILED ***
300 was not greater than 300 (MetricsFailureInjectionSuite.scala:...)
```
Diagnosis (repeated runs on a macOS arm64 runner, capturing the metric
values and scheduler
markers):
- The test **passes 10/10 when run in isolation**, but flakes in the full
suite - so it is a
cross-test/scheduling interaction, not a problem with the query itself.
- On a failing run the metrics are `stage1=300 stage2=5` (exactly the
no-recompute values) and
the case finishes in ~370ms versus ~860ms on a passing run, and the
test's own shuffle gets
**no injected `FetchFailed`**.
`INJECT_SHUFFLE_FETCH_FAILURES` corrupts mapper-0 of the shuffle map stage,
but whether the
downstream reducer observes the resulting `FetchFailed` - and thus forces
the stage-1 recompute
that makes the raw `stage1Metric` exceed 300 - depends on task scheduling
within the shared
`SparkContext`. It occasionally does not fire, leaving `stage1Metric` at
exactly 300.
Retrying until the injection fires makes the test robust without weakening
what it verifies (it
still requires a real recompute), and without touching the shared
`DAGScheduler` injection
machinery that other suites (`SQLLastAttemptMetric*`, DSv2 DML metric
tests) rely on.
### Does this PR introduce _any_ user-facing change?
No. Test-only.
### How was this patch tested?
Ran the **full** `MetricsFailureInjectionSuite` **8×** on a `macos-15`
runner (where it was ~1/6
flaky before): all 8 iterations green (`Tests: succeeded 13, failed 0`).
- Before (flaky on `macos-26`):
https://github.com/apache/spark/actions/runs/28753698265
- After (this fix, 8/8 green on fork):
https://github.com/HyukjinKwon/spark/actions/runs/28780271388
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code
Closes #57037 from HyukjinKwon/ci-fix/agent4-metrics-injection-flake-pr.
Authored-by: Hyukjin Kwon <[email protected]>
Signed-off-by: Hyukjin Kwon <[email protected]>
(cherry picked from commit 09f08e2ffec552b9610507ef827bded21a0e0dce)
Signed-off-by: Hyukjin Kwon <[email protected]>
---
.../metric/MetricsFailureInjectionSuite.scala | 70 +++++++++++++++-------
1 file changed, 48 insertions(+), 22 deletions(-)
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/execution/metric/MetricsFailureInjectionSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/execution/metric/MetricsFailureInjectionSuite.scala
index 847a12f4f305..e3f2e7ca2f08 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/execution/metric/MetricsFailureInjectionSuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/execution/metric/MetricsFailureInjectionSuite.scala
@@ -316,28 +316,54 @@ class MetricsFailureInjectionSuite
setUpTestTable("test_table")
withSparkContextConf(
config.Tests.INJECT_SHUFFLE_FETCH_FAILURES.key ->
injectFailure.toString) {
- val stage1MetricsExpr = incrementMetrics(Seq(stage1Metric,
stage1SLAMetric))
- val udfRand =
- udf {
- () => {
- new Random().nextDouble()
- }
- }.asNondeterministic().apply().expr
- val stage1 = spark.read.table("test_table")
- .withColumn("non_deterministic_col", Column(udfRand))
- .filter(Column(stage1MetricsExpr))
- val stage2MetricsExpr = incrementMetrics(Seq(stage2Metric,
stage2SLAMetric))
- val stage2 = stage1
- .groupBy("low_cardinality_col")
- .avg("non_deterministic_col")
- .filter(Column(stage2MetricsExpr))
- // Add an extra stage with a single task to avoid flaky failures. If
a ResultTask
- // returns non-deterministic results to the client, it forces the
query to abort
- // instead of retrying the input stages.
- val finalDf = stage2.repartition(1).as[(Int, Double)]
- val result = finalDf.collect()
- // Don't compare the second value, since it's random.
- assert(result.map(_._1).toSet === (0 until 5).toSet)
+ def runOnce(): Dataset[_] = {
+ val stage1MetricsExpr = incrementMetrics(Seq(stage1Metric,
stage1SLAMetric))
+ val udfRand =
+ udf {
+ () => {
+ new Random().nextDouble()
+ }
+ }.asNondeterministic().apply().expr
+ val stage1 = spark.read.table("test_table")
+ .withColumn("non_deterministic_col", Column(udfRand))
+ .filter(Column(stage1MetricsExpr))
+ val stage2MetricsExpr = incrementMetrics(Seq(stage2Metric,
stage2SLAMetric))
+ val stage2 = stage1
+ .groupBy("low_cardinality_col")
+ .avg("non_deterministic_col")
+ .filter(Column(stage2MetricsExpr))
+ // Add an extra stage with a single task to avoid flaky failures.
If a ResultTask
+ // returns non-deterministic results to the client, it forces the
query to abort
+ // instead of retrying the input stages.
+ val finalDf = stage2.repartition(1).as[(Int, Double)]
+ val result = finalDf.collect()
+ // Don't compare the second value, since it's random.
+ assert(result.map(_._1).toSet === (0 until 5).toSet)
+ finalDf
+ }
+
+ // The INJECT_SHUFFLE_FETCH_FAILURES machinery corrupts mapper-0 of
the first successful
+ // attempt of the shuffle map stage. Whether the downstream reducer
observes the resulting
+ // FetchFailed (and thus forces the stage-1 recompute that inflates
the raw metric) depends
+ // on task scheduling within the shared SparkContext; across the
suite it occasionally does
+ // not fire, leaving stage1Metric at exactly 300 and failing "value
> 300" (a ~1/6 flake,
+ // more frequent on slower runners such as macOS arm64). When we
require a recompute
+ // (injectFailure = true), re-run the query until the injection
actually fires. Each attempt
+ // resets the metrics, so a successful attempt is indistinguishable
from a first-try success.
+ var finalDf = runOnce()
+ if (injectFailure) {
+ var attempts = 1
+ while (stage1Metric.value <= 300 && attempts < 10) {
+ stage1Metric.reset()
+ stage2Metric.reset()
+ stage1SLAMetric.reset()
+ stage2SLAMetric.reset()
+ finalDf = runOnce()
+ attempts += 1
+ }
+ assert(stage1Metric.value > 300,
+ s"fetch-failure injection did not force a recompute after
$attempts attempts")
+ }
postRunChecks(finalDf)
stage1Metric.reset()
stage2Metric.reset()
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
