[SPARK-41187][CORE] LiveExecutor MemoryLeak in AppStatusListener when ExecutorLost happen #38702

wineternity · 2022-11-18T03:37:24Z

What changes were proposed in this pull request?

Ignore the SparkListenerTaskEnd with Reason "Resubmitted" in AppStatusListener to avoid memory leak

Why are the changes needed?

For a long running spark thriftserver, LiveExecutor will be accumulated in the deadExecutors HashMap and cause message event queue processing slowly.
For a every task, actually always sent out a SparkListenerTaskStart event and a SparkListenerTaskEnd event, they are always pairs. But in a executor lost situation, it send out event like following steps.

a) There was a pair of task start and task end event which were fired for the task (let us call it Tr)
b) When executor which ran Tr was lost, while stage is still running, a task end event with reason Resubmitted is fired for Tr.
c) Subsequently, a new task start and task end will be fired for the retry of Tr.

The processing of the Resubmitted task end event in AppStatusListener can lead to negative LiveStage.activeTasks since there's no corresponding SparkListenerTaskStart event for each of them. The negative activeTasks will make the stage always remains in the live stage list as it can never meet the condition activeTasks == 0. This in turn causes the dead executor to never be cleaned up if that live stage's submissionTime is less than the dead executor's removeTime( see isExecutorActiveForLiveStages). Since this kind of SparkListenerTaskEnd is useless here, we simply ignore it.

Check SPARK-41187 for evidences.

Does this PR introduce any user-facing change?

No

How was this patch tested?

New UT Added
Test in thriftserver env

The way to reproduce

I try to reproduce it in spark shell, but it is a little bit handy

start spark-shell , set spark.dynamicAllocation.maxExecutors=2 for convient
bin/spark-shell --driver-java-options "-agentlib:jdwp=transport=dt_socket,server=y,suspend=n,address=8006"
run a job with shuffle
sc.parallelize(1 to 1000, 10).map { x => Thread.sleep(1000) ; (x % 3, x) }.reduceByKey((a, b) => a + b).collect()
After some ShuffleMapTask finished, kill one or two executor to let tasks resubmitted
check by heap dump or debug or log

itholic · 2022-11-18T05:26:49Z

Can we change the JIRA format in the title such as "[SPARK-41187][CORE] ...".

Check the Spark contribution guide also would helpful!

AmplabJenkins · 2022-11-18T22:47:05Z

Can one of the admins verify this patch?

wineternity · 2022-11-21T09:42:31Z

@itholic @vanzin Could you help review this patch? thanks very much.

wineternity · 2022-11-25T06:03:07Z

cc @cloud-fan

core/src/test/scala/org/apache/spark/status/AppStatusListenerSuite.scala

core/src/main/scala/org/apache/spark/status/AppStatusListener.scala

cloud-fan · 2022-11-29T08:54:36Z

cc @jiangxb1987 @Ngone51

core/src/test/scala/org/apache/spark/status/AppStatusListenerSuite.scala

core/src/main/scala/org/apache/spark/status/AppStatusListener.scala

mridulm

The change looks good to me.
+CC @Ngone51

Btw, do you also want to remove the if (event.taskInfo == null) { check in beginning of onTaskEnd ?

Make it a precondition check ? Preconditions.checkNotNull(event.taskInfo)

wineternity · 2022-12-07T06:34:40Z

The change looks good to me. +CC @Ngone51

Btw, do you also want to remove the if (event.taskInfo == null) { check in beginning of onTaskEnd ?

Make it a precondition check ? Preconditions.checkNotNull(event.taskInfo)

Yes, it can be change to a precondition check. Maybe I can change it in a new pr after test.

core/src/main/scala/org/apache/spark/status/AppStatusListener.scala

Ngone51 · 2022-12-09T03:45:10Z

core/src/test/scala/org/apache/spark/status/AppStatusListenerSuite.scala

@@ -1849,6 +1849,68 @@ abstract class AppStatusListenerSuite extends SparkFunSuite with BeforeAndAfter
    checkInfoPopulated(listener, logUrlMap, processId)
  }

+  test(s"SPARK-41187: Stage should be removed from liveStages to avoid deadExecutors accumulated") {


Suggested change

test(s"SPARK-41187: Stage should be removed from liveStages to avoid deadExecutors accumulated") {

test("SPARK-41187: Stage should be removed from liveStages to avoid deadExecutors accumulated") {

Ngone51 · 2022-12-09T03:51:49Z

Btw, do you also want to remove the if (event.taskInfo == null) { check in beginning of onTaskEnd ?

@mridulm Since the latest PR fix doesn't involve the metrics, I think we can skip this removal to keep the current changes as much simpler as possible. We can back to it when working on metrics stuff.

core/src/test/scala/org/apache/spark/status/AppStatusListenerSuite.scala

core/src/main/scala/org/apache/spark/status/AppStatusListener.scala

…ExecutorLost happen Co-authored-by: wuyi <yi.wu@databricks.com>

dongjoon-hyun

+1, LGTM. Thank you, @wineternity and all.

wineternity · 2022-12-11T09:22:35Z

+1, LGTM. Thank you, @wineternity and all.

my pleasure

… ExecutorLost happen ### What changes were proposed in this pull request? Ignore the SparkListenerTaskEnd with Reason "Resubmitted" in AppStatusListener to avoid memory leak ### Why are the changes needed? For a long running spark thriftserver, LiveExecutor will be accumulated in the deadExecutors HashMap and cause message event queue processing slowly. For a every task, actually always sent out a `SparkListenerTaskStart` event and a `SparkListenerTaskEnd` event, they are always pairs. But in a executor lost situation, it send out event like following steps. a) There was a pair of task start and task end event which were fired for the task (let us call it Tr) b) When executor which ran Tr was lost, while stage is still running, a task end event with reason `Resubmitted` is fired for Tr. c) Subsequently, a new task start and task end will be fired for the retry of Tr. The processing of the `Resubmitted` task end event in AppStatusListener can lead to negative `LiveStage.activeTasks` since there's no corresponding `SparkListenerTaskStart` event for each of them. The negative activeTasks will make the stage always remains in the live stage list as it can never meet the condition activeTasks == 0. This in turn causes the dead executor to never be cleaned up if that live stage's submissionTime is less than the dead executor's removeTime( see isExecutorActiveForLiveStages). Since this kind of `SparkListenerTaskEnd` is useless here, we simply ignore it. Check [SPARK-41187](https://issues.apache.org/jira/browse/SPARK-41187) for evidences. ### Does this PR introduce _any_ user-facing change? No ### How was this patch tested? New UT Added Test in thriftserver env ### The way to reproduce I try to reproduce it in spark shell, but it is a little bit handy 1. start spark-shell , set spark.dynamicAllocation.maxExecutors=2 for convient ` bin/spark-shell --driver-java-options "-agentlib:jdwp=transport=dt_socket,server=y,suspend=n,address=8006"` 2. run a job with shuffle `sc.parallelize(1 to 1000, 10).map { x => Thread.sleep(1000) ; (x % 3, x) }.reduceByKey((a, b) => a + b).collect()` 3. After some ShuffleMapTask finished, kill one or two executor to let tasks resubmitted 4. check by heap dump or debug or log Closes #38702 from wineternity/SPARK-41187. Authored-by: yuanyimeng <yuanyimeng@youzan.com> Signed-off-by: Mridul Muralidharan <mridul<at>gmail.com> (cherry picked from commit 7e7bc94) Signed-off-by: Mridul Muralidharan <mridulatgmail.com>

mridulm · 2022-12-12T04:51:21Z

Merged to master and brach-3.3
Thanks for working on this @wineternity !
Thanks for the reviews @cloud-fan, @Ngone51, @dongjoon-hyun :-)

… ExecutorLost happen ### What changes were proposed in this pull request? Ignore the SparkListenerTaskEnd with Reason "Resubmitted" in AppStatusListener to avoid memory leak ### Why are the changes needed? For a long running spark thriftserver, LiveExecutor will be accumulated in the deadExecutors HashMap and cause message event queue processing slowly. For a every task, actually always sent out a `SparkListenerTaskStart` event and a `SparkListenerTaskEnd` event, they are always pairs. But in a executor lost situation, it send out event like following steps. a) There was a pair of task start and task end event which were fired for the task (let us call it Tr) b) When executor which ran Tr was lost, while stage is still running, a task end event with reason `Resubmitted` is fired for Tr. c) Subsequently, a new task start and task end will be fired for the retry of Tr. The processing of the `Resubmitted` task end event in AppStatusListener can lead to negative `LiveStage.activeTasks` since there's no corresponding `SparkListenerTaskStart` event for each of them. The negative activeTasks will make the stage always remains in the live stage list as it can never meet the condition activeTasks == 0. This in turn causes the dead executor to never be cleaned up if that live stage's submissionTime is less than the dead executor's removeTime( see isExecutorActiveForLiveStages). Since this kind of `SparkListenerTaskEnd` is useless here, we simply ignore it. Check [SPARK-41187](https://issues.apache.org/jira/browse/SPARK-41187) for evidences. ### Does this PR introduce _any_ user-facing change? No ### How was this patch tested? New UT Added Test in thriftserver env ### The way to reproduce I try to reproduce it in spark shell, but it is a little bit handy 1. start spark-shell , set spark.dynamicAllocation.maxExecutors=2 for convient ` bin/spark-shell --driver-java-options "-agentlib:jdwp=transport=dt_socket,server=y,suspend=n,address=8006"` 2. run a job with shuffle `sc.parallelize(1 to 1000, 10).map { x => Thread.sleep(1000) ; (x % 3, x) }.reduceByKey((a, b) => a + b).collect()` 3. After some ShuffleMapTask finished, kill one or two executor to let tasks resubmitted 4. check by heap dump or debug or log Closes apache#38702 from wineternity/SPARK-41187. Authored-by: yuanyimeng <yuanyimeng@youzan.com> Signed-off-by: Mridul Muralidharan <mridul<at>gmail.com>

github-actions bot added the CORE label Nov 18, 2022

wineternity changed the title ~~SPARK-41187 [Core] LiveExecutor MemoryLeak in AppStatusListener when ExecutorLost happen~~ [SPARK-41187][Core] LiveExecutor MemoryLeak in AppStatusListener when ExecutorLost happen Nov 18, 2022

HyukjinKwon changed the title ~~[SPARK-41187][Core] LiveExecutor MemoryLeak in AppStatusListener when ExecutorLost happen~~ [SPARK-41187][CORE] LiveExecutor MemoryLeak in AppStatusListener when ExecutorLost happen Nov 22, 2022

cloud-fan reviewed Nov 25, 2022

View reviewed changes

core/src/test/scala/org/apache/spark/status/AppStatusListenerSuite.scala Show resolved Hide resolved

cloud-fan reviewed Nov 29, 2022

View reviewed changes

core/src/main/scala/org/apache/spark/status/AppStatusListener.scala Outdated Show resolved Hide resolved

Ngone51 reviewed Nov 29, 2022

View reviewed changes

Ngone51 reviewed Nov 30, 2022

View reviewed changes

core/src/main/scala/org/apache/spark/status/AppStatusListener.scala Outdated Show resolved Hide resolved

mridulm reviewed Nov 30, 2022

View reviewed changes

core/src/main/scala/org/apache/spark/status/AppStatusListener.scala Outdated Show resolved Hide resolved

core/src/main/scala/org/apache/spark/status/AppStatusListener.scala Outdated Show resolved Hide resolved

wineternity force-pushed the SPARK-41187 branch from 0ff13b8 to 0476a76 Compare December 6, 2022 13:34

mridulm approved these changes Dec 7, 2022

View reviewed changes

Ngone51 reviewed Dec 9, 2022

View reviewed changes

core/src/main/scala/org/apache/spark/status/AppStatusListener.scala Outdated Show resolved Hide resolved

Ngone51 reviewed Dec 9, 2022

View reviewed changes

core/src/test/scala/org/apache/spark/status/AppStatusListenerSuite.scala Outdated Show resolved Hide resolved

wineternity force-pushed the SPARK-41187 branch 2 times, most recently from 634eefe to ea1307c Compare December 9, 2022 09:08

Ngone51 reviewed Dec 9, 2022

View reviewed changes

core/src/main/scala/org/apache/spark/status/AppStatusListener.scala Outdated Show resolved Hide resolved

Ngone51 approved these changes Dec 9, 2022

View reviewed changes

SPARK-41187 [Core] LiveExecutor MemoryLeak in AppStatusListener when …

e8e2318

…ExecutorLost happen Co-authored-by: wuyi <yi.wu@databricks.com>

wineternity force-pushed the SPARK-41187 branch from ea1307c to e8e2318 Compare December 9, 2022 15:59

dongjoon-hyun approved these changes Dec 10, 2022

View reviewed changes

mridulm closed this in 7e7bc94 Dec 12, 2022

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[SPARK-41187][CORE] LiveExecutor MemoryLeak in AppStatusListener when ExecutorLost happen #38702

[SPARK-41187][CORE] LiveExecutor MemoryLeak in AppStatusListener when ExecutorLost happen #38702

wineternity commented Nov 18, 2022 •

edited

itholic commented Nov 18, 2022

AmplabJenkins commented Nov 18, 2022

wineternity commented Nov 21, 2022

wineternity commented Nov 25, 2022

cloud-fan commented Nov 29, 2022

mridulm left a comment

wineternity commented Dec 7, 2022

Ngone51 Dec 9, 2022

Ngone51 commented Dec 9, 2022

dongjoon-hyun left a comment •

edited

wineternity commented Dec 11, 2022

mridulm commented Dec 12, 2022

	test(s"SPARK-41187: Stage should be removed from liveStages to avoid deadExecutors accumulated") {
	test("SPARK-41187: Stage should be removed from liveStages to avoid deadExecutors accumulated") {

[SPARK-41187][CORE] LiveExecutor MemoryLeak in AppStatusListener when ExecutorLost happen #38702

[SPARK-41187][CORE] LiveExecutor MemoryLeak in AppStatusListener when ExecutorLost happen #38702

Conversation

wineternity commented Nov 18, 2022 • edited

What changes were proposed in this pull request?

Why are the changes needed?

Does this PR introduce any user-facing change?

How was this patch tested?

The way to reproduce

itholic commented Nov 18, 2022

AmplabJenkins commented Nov 18, 2022

wineternity commented Nov 21, 2022

wineternity commented Nov 25, 2022

cloud-fan commented Nov 29, 2022

mridulm left a comment

Choose a reason for hiding this comment

wineternity commented Dec 7, 2022

Ngone51 Dec 9, 2022

Choose a reason for hiding this comment

Ngone51 commented Dec 9, 2022

dongjoon-hyun left a comment • edited

Choose a reason for hiding this comment

wineternity commented Dec 11, 2022

mridulm commented Dec 12, 2022

wineternity commented Nov 18, 2022 •

edited

dongjoon-hyun left a comment •

edited