This is an automated email from the ASF dual-hosted git repository.
cloud-fan pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new 41cc3049457c [SPARK-56773][CORE][TEST][FOLLOWUP] Always allocate
INJECT_SHUFFLE_FETCH_FAILURES injection maps
41cc3049457c is described below
commit 41cc3049457cfbd74f9717b2ada9a9040eb5e350
Author: Wenchen Fan <[email protected]>
AuthorDate: Sun Jun 21 20:50:37 2026 -0700
[SPARK-56773][CORE][TEST][FOLLOWUP] Always allocate
INJECT_SHUFFLE_FETCH_FAILURES injection maps
### What changes were proposed in this pull request?
Followup to https://github.com/apache/spark/pull/55738.
The three `INJECT_SHUFFLE_FETCH_FAILURES` injection-state maps in
`DAGScheduler` (`injectShuffleFetchFailuresCorruptedAttempt`,
`injectShuffleFetchFailuresPendingDelayedCorruption`,
`injectShuffleFetchFailuresDownstreamSuccessCount`) are initialized with `if
(Utils.isTesting) new ConcurrentHashMap else null`. This patch allocates them
unconditionally instead.
### Why are the changes needed?
The cleanup site in `cleanupStateForJobAndIndependentStages` (and the other
use-sites) re-evaluate `if (Utils.isTesting)` to decide whether to dereference
these maps. `Utils.isTesting` reads the mutable `spark.testing` system
property, so it can return a different value when the `DAGScheduler` is
constructed than at the later use-sites. If it does, a `null` map is
dereferenced and the `DAGScheduler` event loop crashes with a
`NullPointerException`. Allocating the maps unconditionally [...]
### Does this PR introduce _any_ user-facing change?
No. The maps are only ever populated inside the config-gated
`INJECT_SHUFFLE_FETCH_FAILURES` test paths, so in production they stay empty
and there is no behavioral change beyond allocating an empty map.
### How was this patch tested?
Existing tests. This is a defensive initialization change; the maps stay
empty unless the existing `INJECT_SHUFFLE_FETCH_FAILURES` test paths populate
them, so the existing coverage exercises them unchanged.
### Was this patch authored or co-authored using generative AI tooling?
Yes, drafted with Claude Code (Anthropic).
Closes #56631 from cloud-fan/SPARK-56773-followup.
Authored-by: Wenchen Fan <[email protected]>
Signed-off-by: Wenchen Fan <[email protected]>
---
.../scala/org/apache/spark/scheduler/DAGScheduler.scala | 14 +++++++++++---
1 file changed, 11 insertions(+), 3 deletions(-)
diff --git a/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
b/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
index 0754a90459d5..4a0db18d328f 100644
--- a/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
+++ b/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
@@ -179,12 +179,20 @@ private[spark] class DAGScheduler(
private[spark] val jobIdToQueryExecutionId = new ConcurrentHashMap[Int,
java.lang.Long]()
+ // The maps below back the test-only INJECT_SHUFFLE_FETCH_FAILURES
machinery. They are always
+ // allocated rather than gated on `Utils.isTesting`: that helper reads the
mutable
+ // `spark.testing` system property, so it can return a different value when
this DAGScheduler is
+ // constructed than at the later use-sites. A construction-time `else null`
would then be
+ // dereferenced by a use-site that re-checks `Utils.isTesting` and sees
`true`, throwing an NPE
+ // that crashes the event loop. The maps are only ever populated inside the
config-gated test
+ // paths, so in production they stay empty and carry no behavioral cost
beyond an empty map.
+
// For INJECT_SHUFFLE_FETCH_FAILURES: per-shuffleId, the stage attempt whose
partition-0 task
// we corrupted. Read to (a) avoid re-corrupting that partition on
recompute, and (b) decide
// when to fire INJECT_SHUFFLE_FORCE_CHECKSUM_MISMATCH_ON_RECOMPUTE - the
recompute is the
// task whose stageAttemptId is not the recorded one.
private val injectShuffleFetchFailuresCorruptedAttempt:
ConcurrentHashMap[Int, Int] =
- if (Utils.isTesting) new ConcurrentHashMap[Int, Int]() else null
+ new ConcurrentHashMap[Int, Int]()
// For INJECT_SHUFFLE_FETCH_FAILURES_DOWNSTREAM_DELAY > 0: shuffles whose
mapper-0 corruption
// has been deferred until enough downstream consumer tasks succeed. The
value is the mapId
@@ -193,12 +201,12 @@ private[spark] class DAGScheduler(
// is still a real one (only the executorId is INVALID_EXECUTOR_ID).
private val injectShuffleFetchFailuresPendingDelayedCorruption
: ConcurrentHashMap[Int, (Long, BlockManagerId)] =
- if (Utils.isTesting) new ConcurrentHashMap[Int, (Long, BlockManagerId)]()
else null
+ new ConcurrentHashMap[Int, (Long, BlockManagerId)]()
// For INJECT_SHUFFLE_FETCH_FAILURES_DOWNSTREAM_DELAY: per-shuffle counter
of consumer
// task-success events observed so far.
private val injectShuffleFetchFailuresDownstreamSuccessCount:
ConcurrentHashMap[Int, Int] =
- if (Utils.isTesting) new ConcurrentHashMap[Int, Int]() else null
+ new ConcurrentHashMap[Int, Int]()
// Build the bogus BlockManagerId used by INJECT_SHUFFLE_FETCH_FAILURES to
mark a corrupted
// MapStatus: keeps the original host/port/topology so the consumer's
locality preference
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
