Ziqi Liu created SPARK-59845:
--------------------------------
Summary: Register an accumulator with the TaskContext only after
it is fully deserialized
Key: SPARK-59845
URL: https://issues.apache.org/jira/browse/SPARK-59845
Project: Spark
Issue Type: Bug
Components: Spark Core
Affects Versions: 4.3.0
Reporter: Ziqi Liu
{{AccumulatorV2.readObject}} registers the accumulator with the
{{{}TaskContext{}}}, which publishes it where other threads can reach it. That
runs at {{{}AccumulatorV2{}}}'s own slot during deserialization, and Java
serialization reads class data superclass-first, so no subclass field has been
assigned yet.
A subclass holding its state in a {{val}} therefore exposes a null for that
window. The reader that hits it in practice is the executor heartbeater, which
calls {{isZero}} on every registered accumulator every
{{{}spark.executor.heartbeatInterval{}}}. The NPE is thrown on the
{{executor-heartbeater}} thread and propagates out of a {{scheduleAtFixedRate}}
task, so the executor stops heartbeating permanently; the driver then removes
it on heartbeat timeout, losing anything held only there — e.g. local
checkpoint blocks, which surfaces as {{{}CHECKPOINT_RDD_BLOCK_ID_NOT_FOUND{}}}.
Because a Scala {{val}} compiles to a final field, there is a second aspect:
the reflective write happens after publication, so per JLS 17.5.3 a reader has
no guarantee of ever seeing it and the JIT may fold the read.
SPARK-20977 fixed this locally in {{CollectionAccumulator}} by making its state
a non-final field reached through a lazily-initializing accessor. Every
subclass since has had to re-derive that workaround, and not all have:
{{SetAccumulator}} in {{sql/execution/debug}} still dereferences a {{private
val _set}} from {{{}isZero{}}}, and {{EventTimeStatsAccum}} avoids the NPE only
incidentally, because Scala's {{==}} is null-safe — it reports the wrong answer
and ships a null instead.
Fix: register in {{{}readResolve{}}}, which Java serialization invokes only
after every class in the hierarchy has been read. The accumulator is then never
observable half-initialized and no subclass needs to guard its own state.
Verified with a new test using a subclass that holds state in a plain {{val}}
with no null handling: it fails with the NPE on current {{master}} and passes
with registration moved, while the existing
{{{}LongAccumulator{}}}/{{{}DoubleAccumulator{}}}/{{{}CollectionAccumulator{}}}
tests are unaffected.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
