[spark] branch master updated: [SPARK-36798][CORE] Wait for listeners to finish before flushing metrics

[mridulm80]

This is an automated email from the ASF dual-hosted git repository.

mridulm80 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git


The following commit(s) were added to refs/heads/master by this push:
     new d758ae3  [SPARK-36798][CORE] Wait for listeners to finish before 
flushing metrics
d758ae3 is described below

commit d758ae3164933039c9c4766d8085ba3003df7b49
Author: BOOTMGR <panchal.hars...@gmail.com>
AuthorDate: Thu Oct 7 08:08:35 2021 -0500

    [SPARK-36798][CORE] Wait for listeners to finish before flushing metrics
    
    ### What changes were proposed in this pull request?
    When `SparkContext` is shutting down, wait for listener bus to finish and 
then only flush `MetricsSystem`.
    
    ### Why are the changes needed?
    In current implementation, when `SparkContext.stop()` is called, 
`metricsSystem.report()` is called before `listenerBus.stop()`. In this case, 
if some listener is producing some metrics, they would never reach sink.
    
    ### Does this PR introduce _any_ user-facing change?
    No
    
    ### How was this patch tested?
    NA
    
    Closes #34039 from BOOTMGR/SPARK-36798.
    
    Authored-by: BOOTMGR <panchal.hars...@gmail.com>
    Signed-off-by: Mridul Muralidharan <mridul<at>gmail.com>
---
 core/src/main/scala/org/apache/spark/SparkContext.scala | 10 +++++-----
 1 file changed, 5 insertions(+), 5 deletions(-)

diff --git a/core/src/main/scala/org/apache/spark/SparkContext.scala 
b/core/src/main/scala/org/apache/spark/SparkContext.scala
index e27499a15..fab46d5 100644
--- a/core/src/main/scala/org/apache/spark/SparkContext.scala
+++ b/core/src/main/scala/org/apache/spark/SparkContext.scala
@@ -2080,11 +2080,6 @@ class SparkContext(config: SparkConf) extends Logging {
     Utils.tryLogNonFatalError {
       _ui.foreach(_.stop())
     }
-    if (env != null) {
-      Utils.tryLogNonFatalError {
-        env.metricsSystem.report()
-      }
-    }
     Utils.tryLogNonFatalError {
       _cleaner.foreach(_.stop())
     }
@@ -2103,6 +2098,11 @@ class SparkContext(config: SparkConf) extends Logging {
         _listenerBusStarted = false
       }
     }
+    if (env != null) {
+      Utils.tryLogNonFatalError {
+        env.metricsSystem.report()
+      }
+    }
     Utils.tryLogNonFatalError {
       _plugins.foreach(_.shutdown())
     }

---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscr...@spark.apache.org
For additional commands, e-mail: commits-h...@spark.apache.org