[
https://issues.apache.org/jira/browse/SPARK-59845?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
ASF GitHub Bot updated SPARK-59845:
-----------------------------------
Labels: pull-request-available (was: )
> 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
> Priority: Major
> Labels: pull-request-available
>
> {{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]
