[SPARK-25910][CORE] accumulator updates from previous stage attempt should not log error

[cloud-fan]

What changes were proposed in this pull request?

For shuffle map stages, we may have multiple attempts, while only the latest attempt is active. However, the scheduler still accepts successful tasks from previous attempts, to speed up the execution.

Each stage attempt has a StageInfo instance, which contains TaskMetrics. TaskMetrics has a bunch of accumulators to track the metrics like CPU time, etc. However, a stage only keeps the StageInfo of the latest attempt, which means the StageInfo of previous attempts will be GCed, and their accumulators of TaskMetrics will be cleaned.

This causes a problem: When the scheduler accepts a successful task from a previous attempt, and tries to update accumulators, we may fail to get the accumulators from AccumulatorContext, as they are already cleaned. And we may hit error log like

18/10/21 15:30:24 INFO ContextCleaner: Cleaned accumulator 2868 (name: internal.metrics.executorDeserializeTime)
18/10/21 15:30:24 ERROR DAGScheduler: Failed to update accumulators for task 7927
org.apache.spark.SparkException: attempted to access non-existent accumulator 2868
at org.apache.spark.scheduler.DAGScheduler$$anonfun$updateAccumulators$1.apply(DAGScheduler.scala:1267)
...

This PR proposes a simple fix: When the scheduler receives successful tasks from previous attempts, don't update accumulators. Accumulators of previous stage attemps are not tracked anymore, so we don't need to update them.

How was this patch tested?

a new test.

[cloud-fan]

cc @vanzin @zsxwing @jiangxb1987

[SparkQA]

Test build #98355 has finished for PR 22923 at commit 4d9cbe0.

• This patch fails SparkR unit tests.
• This patch merges cleanly.
• This patch adds no public classes.

[HyukjinKwon]

retest this please

[SparkQA]

Test build #98416 has finished for PR 22923 at commit 4d9cbe0.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.

[jiangxb1987]

LGTM

[zsxwing]

We need to always update user accumulators. Right now such task metrics just cause some annoying error logs, seems not worth to fix.

[cloud-fan]

We need to always update user accumulators

Ah that's a good point.

I'm going to close it. I missed one thing: the AppStatusListener will keep the StageInfo instance until all tasks of that stage attempt is finished. See #22209

So this bug should already have been fixed.