[SPARK-29399][CORE] Remove old ExecutorPlugin interface

SPARK-29397 added new interfaces for creating driver and executor
plugins. These were added in a new, more isolated package that does
not pollute the main o.a.s package.

The old interface is now redundant. Since it's a DeveloperApi and
we're about to have a new major release, let's remove it instead of
carrying more baggage forward.

Closes apache#26390 from vanzin/SPARK-29399.

Authored-by: Marcelo Vanzin <[email protected]>
Signed-off-by: HyukjinKwon <[email protected]>

Author: Marcelo Vanzin

core/src/main/java/org/apache/spark/ExecutorPlugin.java

@@ -24,7 +24,7 @@
  * A plugin that can be dynamically loaded into a Spark application.
  * <p>
  * Plugins can be loaded by adding the plugin's class name to the appropriate Spark configuration.
- * Check the Spark configuration documentation for details.
+ * Check the Spark monitoring guide for details.
  * <p>
  * Plugins have two optional components: a driver-side component, of which a single instance is
  * created per application, inside the Spark driver. And an executor-side component, of which one

core/src/main/java/org/apache/spark/ExecutorPluginContext.java

@@ -619,7 +619,9 @@ private[spark] object SparkConf extends Logging {
         "Not used anymore. Please use spark.shuffle.service.index.cache.size"),
       DeprecatedConfig("spark.yarn.credentials.file.retention.count", "2.4.0", "Not used anymore."),
       DeprecatedConfig("spark.yarn.credentials.file.retention.days", "2.4.0", "Not used anymore."),
-      DeprecatedConfig("spark.yarn.services", "3.0.0", "Feature no longer available.")
+      DeprecatedConfig("spark.yarn.services", "3.0.0", "Feature no longer available."),
+      DeprecatedConfig("spark.executor.plugins", "3.0.0",
+        "Feature replaced with new plugin API. See Monitoring documentation.")
     )
 
     Map(configs.map { cfg => (cfg.key -> cfg) } : _*)

core/src/main/java/org/apache/spark/api/plugin/SparkPlugin.java

@@ -137,35 +137,6 @@ private[spark] class Executor(
   // for fetching remote cached RDD blocks, so need to make sure it uses the right classloader too.
   env.serializerManager.setDefaultClassLoader(replClassLoader)
 
-  private val executorPlugins: Seq[ExecutorPlugin] = {
-    val pluginNames = conf.get(EXECUTOR_PLUGINS)
-    if (pluginNames.nonEmpty) {
-      logInfo(s"Initializing the following plugins: ${pluginNames.mkString(", ")}")
-
-      // Plugins need to load using a class loader that includes the executor's user classpath
-      val pluginList: Seq[ExecutorPlugin] =
-        Utils.withContextClassLoader(replClassLoader) {
-          val plugins = Utils.loadExtensions(classOf[ExecutorPlugin], pluginNames, conf)
-          plugins.foreach { plugin =>
-            val pluginSource = new ExecutorPluginSource(plugin.getClass().getSimpleName())
-            val pluginContext = new ExecutorPluginContext(pluginSource.metricRegistry, conf,
-              executorId, executorHostname, isLocal)
-            plugin.init(pluginContext)
-            logInfo("Successfully loaded plugin " + plugin.getClass().getCanonicalName())
-            if (pluginSource.metricRegistry.getNames.size() > 0) {
-              env.metricsSystem.registerSource(pluginSource)
-            }
-          }
-          plugins
-        }
-
-      logInfo("Finished initializing plugins")
-      pluginList
-    } else {
-      Nil
-    }
-  }
-
   // Plugins need to load using a class loader that includes the executor's user classpath
   private val plugins: Option[PluginContainer] = Utils.withContextClassLoader(replClassLoader) {
     PluginContainer(env)
@@ -295,14 +266,6 @@ private[spark] class Executor(
 
     // Notify plugins that executor is shutting down so they can terminate cleanly
     Utils.withContextClassLoader(replClassLoader) {
-      executorPlugins.foreach { plugin =>
-        try {
-          plugin.shutdown()
-        } catch {
-          case e: Exception =>
-            logWarning("Plugin " + plugin.getClass().getCanonicalName() + " shutdown failed", e)
-        }
-      }
       plugins.foreach(_.shutdown())
     }
     if (!isLocal) {

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

@@ -1193,16 +1193,6 @@ package object config {
       .toSequence
       .createWithDefault(Nil)
 
-  private[spark] val EXECUTOR_PLUGINS =
-    ConfigBuilder("spark.executor.plugins")
-      .doc("Comma-separated list of class names for \"plugins\" implementing " +
-        "org.apache.spark.ExecutorPlugin.  Plugins have the same privileges as any task " +
-        "in a Spark executor.  They can also interfere with task execution and fail in " +
-        "unexpected ways.  So be sure to only use this for trusted plugins.")
-      .stringConf
-      .toSequence
-      .createWithDefault(Nil)
-
   private[spark] val CLEANER_PERIODIC_GC_INTERVAL =
     ConfigBuilder("spark.cleaner.periodicGC.interval")
       .timeConf(TimeUnit.SECONDS)