Remove SPARK_EXTRA_LISTENERS environment variable.

Author: JoshRosen

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

@@ -1494,19 +1494,10 @@ class SparkContext(config: SparkConf) extends Logging with ExecutorAllocationCli
     if (listenerBus.hasBeenStarted) {
       throw new IllegalStateException("listener bus has already been started")
     }
-    // Use reflection to instantiate listeners specified via the `spark.extraListeners`
-    // configuration or the SPARK_EXTRA_LISTENERS environment variable. The purpose of
-    // SPARK_EXTRA_LISTENERS is to allow the execution environment to inject custom listeners
-    // without having to worry about them being overridden by user settings in SparkConf.
+    // Use reflection to instantiate listeners specified via `spark.extraListeners`
     try {
-      val listenerClassNames: Seq[String] = {
-        // Merge configurations from both sources
-        val fromSparkConf = conf.get("spark.extraListeners", "").split(',')
-        val fromEnvVar = Option(conf.getenv("SPARK_EXTRA_LISTENERS")).getOrElse("").split(',')
-        // Filter out empty entries, which could occur when overriding environment variables
-        // (e.g. `export SPARK_EXTRA_LISTENERS="foo,$SPARK_EXTRA_LISTENERS`)
-        (fromSparkConf ++ fromEnvVar).map(_.trim).filter(_ != "")
-      }
+      val listenerClassNames: Seq[String] =
+        conf.get("spark.extraListeners", "").split(',').map(_.trim).filter(_ != "")
       for (className <- listenerClassNames) {
         // Use reflection to find the right constructor
         val constructors = {

core/src/test/scala/org/apache/spark/scheduler/SparkListenerSuite.scala

@@ -21,12 +21,11 @@ import java.util.concurrent.Semaphore
 
 import scala.collection.mutable
 
-import org.apache.spark.{LocalSparkContext, SparkConf, SparkContext}
+import org.scalatest.{FunSuite, Matchers}
+
 import org.apache.spark.executor.TaskMetrics
 import org.apache.spark.util.ResetSystemProperties
-
-import org.mockito.Mockito._
-import org.scalatest.{FunSuite, Matchers}
+import org.apache.spark.{LocalSparkContext, SparkConf, SparkContext}
 
 class SparkListenerSuite extends FunSuite with LocalSparkContext with Matchers
   with ResetSystemProperties {
@@ -370,34 +369,6 @@ class SparkListenerSuite extends FunSuite with LocalSparkContext with Matchers
     }.size should be (1)
   }
 
-  test("registering listeners via SPARK_EXTRA_LISTENERS") {
-    val SPARK_EXTRA_LISTENERS = classOf[ListenerThatAcceptsSparkConf].getName + "," +
-      classOf[BasicJobCounter].getName
-    val conf = spy(new SparkConf().setMaster("local").setAppName("test"))
-    when(conf.getenv("SPARK_EXTRA_LISTENERS")).thenReturn(SPARK_EXTRA_LISTENERS)
-    when(conf.clone).thenReturn(conf)  // so that our mock is still used
-    sc = new SparkContext(conf)
-    sc.listenerBus.sparkListeners.collect { case x: BasicJobCounter => x}.size should be (1)
-    sc.listenerBus.sparkListeners.collect {
-      case x: ListenerThatAcceptsSparkConf => x
-    }.size should be (1)
-  }
-
-  test("spark.extraListeners and SPARK_EXTRA_LISTENERS configurations are merged") {
-    // This test ensures that we don't accidentally change the behavior such that one setting
-    // overrides the other:
-    val SPARK_EXTRA_LISTENERS = classOf[ListenerThatAcceptsSparkConf].getName
-    val conf = spy(new SparkConf().setMaster("local").setAppName("test")
-      .set("spark.extraListeners", classOf[BasicJobCounter].getName))
-    when(conf.getenv("SPARK_EXTRA_LISTENERS")).thenReturn(SPARK_EXTRA_LISTENERS)
-    when(conf.clone).thenReturn(conf)  // so that our mock is still used
-    sc = new SparkContext(conf)
-    sc.listenerBus.sparkListeners.collect { case x: BasicJobCounter => x}.size should be (1)
-    sc.listenerBus.sparkListeners.collect {
-      case x: ListenerThatAcceptsSparkConf => x
-    }.size should be (1)
-  }
-
   /**
    * Assert that the given list of numbers has an average that is greater than zero.
    */