Ngone51 commented on a change in pull request #31839:
URL: https://github.com/apache/spark/pull/31839#discussion_r594897153
##########
File path:
sql/core/src/test/scala/org/apache/spark/sql/SparkSessionBuilderSuite.scala
##########
@@ -40,6 +45,37 @@ class SparkSessionBuilderSuite extends SparkFunSuite with
BeforeAndAfterEach {
SparkSession.clearDefaultSession()
}
+ test("SPARK-34087: Fix memory leak of ExecutionListenerBus") {
+ val spark = SparkSession.builder()
+ .master("local")
+ .getOrCreate()
+
+ def listenersNum(): Int = {
+ spark.sparkContext
+ .listenerBus
+ .listeners
+ .asScala
+ .count(_.isInstanceOf[ExecutionListenerBus])
+ }
+
+ (1 to 10).foreach { _ =>
+ spark.cloneSession()
+ SparkSession.clearActiveSession()
+ }
+
+ var num = listenersNum()
+ // Before GC, the number of ExecutionListenerBus is (1 + 10)
+ assert(num === 11)
+ eventually(timeout(10.seconds), interval(1.seconds)) {
+ System.gc()
+ num = listenersNum()
+ // After GC, the number of ExecutionListenerBus should be less than 11
+ // Since GC can't 100% guarantee all out-of-referenced objects be
cleaned at one time,
+ // here, we check at least one listener is cleaned up to prove the
mechanism works.
+ assert(num < 11)
Review comment:
We actually create 11 sessions: original one and 10 cloned.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
