upgrade spark call home service (apache#1029)

Author: Ling Yuan

core/src/main/scala/org/apache/spark/SparkContext.scala

@@ -632,9 +632,8 @@ class SparkContext(config: SparkConf) extends Logging {
       }
     _executorAllocationManager.foreach(_.start())
 
-    if (conf.get(SPARK_CALL_HOME_ENABLED)) {
-      setupSparkCallHomeListener()
-    }
+
+    setupSparkCallHomeListener()
     setupAndStartListenerBus()
     postEnvironmentUpdate()
     postApplicationStart()
@@ -2521,7 +2520,7 @@ class SparkContext(config: SparkConf) extends Logging {
       val listeners = Utils.loadExtensions(classOf[SparkListenerInterface],
         Seq(SPARK_CALL_HOME_LISTENER_CLASS), conf)
       listeners.foreach { listener =>
-        listenerBus.addToSharedQueue(listener)
+        listenerBus.addToQueue(listener, SPARK_CALL_HOME_QUEUE_NAME)
         logInfo(s"Registered Spark Call Home listener ${listener.getClass().getName()}")
       }
     } catch {

core/src/main/scala/org/apache/spark/internal/config/package.scala

@@ -38,7 +38,9 @@ package object config {
   private[spark] val SPARK_TASK_PREFIX = "spark.task"
   private[spark] val LISTENER_BUS_EVENT_QUEUE_PREFIX = "spark.scheduler.listenerbus.eventqueue"
   private[spark] val SPARK_CALL_HOME_LISTENER_CLASS =
-    "com.apple.aci.data.spark.callhome.listener.v1.SparkCallHomeListenerV1"
+    "com.apple.spark.callhome.listener.SparkCallHomeListenerV2"
+  private[spark] val SPARK_CALL_HOME_QUEUE_NAME =
+    "SparkCallHome"
 
   private[spark] val RESOURCES_DISCOVERY_PLUGIN =
     ConfigBuilder("spark.resources.discoveryPlugin")

pom.xml

@@ -193,7 +193,7 @@
     <jpam.version>1.1</jpam.version>
     <selenium.version>3.141.59</selenium.version>
     <htmlunit.version>2.40.0</htmlunit.version>
-    <acispark.callhome.version>0.2.0</acispark.callhome.version>
+    <acispark.callhome.version>0.2.8</acispark.callhome.version>
     <iceberg.version>0.12.0-apple-preview-7</iceberg.version>
     <!--
     Managed up from older version from Avro; sync with jackson-module-paranamer dependency version