Feature/599 in process launcher (#649)

* #599: InProcessLauncher

* Removed comments

* Test fix

* Test fix

* Exclude hadoop

* NoBackendInProcessLauncher

* NoBackendInProcessLauncher Attempt 2

* Attempt 3

* Directly include hortonworks hadoop

* Remove builder.getEffectiveConfig to avoid checking for SPARK_HOME env variable

* Remove launcher confs

* Add comments

* Debug github build

* Revert

* Delete unnecessary file

* Fix tests

* Fix test

* Update readme

Co-authored-by: jozefbakus <[email protected]>

Author: kevinwallimann

README.md

@@ -160,8 +160,6 @@ spark.submitApi=yarn
 
 #Submit api = YARN
 sparkYarnSink.submitTimeout=160000
-sparkYarnSink.hadoopConfDir=/opt/hadoop
-sparkYarnSink.sparkHome=/opt/spark
 sparkYarnSink.master=yarn
 sparkYarnSink.filesToDeploy=
 sparkYarnSink.additionalConfs.spark.ui.port=
@@ -200,6 +198,13 @@ db.skip.liquibase=false
 spring.liquibase.change-log=classpath:/db_scripts/liquibase/db.changelog.yml
 ```
 
+## Tomcat configuration
+The Hadoop configuration directory needs to be added as the environment variable `HADOOP_CONF_DIR` and it has to be added to the web application's classpath.
+
+- The environment variable can be added in `<tomcat-root>/bin/setenv.sh`, e.g. `HADOOP_CONF_DIR=/opt/hadoop`.
+- To add the Hadoop configuration directory to the application classpath, 
+in the file `<tomcat-base>/conf/catalina.properties`, append to the key `shared.loader` the hadoop conf dir, e.g. `shared.loader="/opt/hadoop"`.
+
 ## Embedded Tomcat
 
 For development purposes, hyperdrive-trigger can be executed as an application with an embedded tomcat. Please check out branch **feature/embedded-tomcat-2** to use it.

pom.xml

@@ -61,12 +61,20 @@
         </developer>
     </developers>
 
+    <repositories>
+        <repository>
+            <id>hortonworks</id>
+            <url>https://repo.hortonworks.com/content/repositories/releases/</url>
+        </repository>
+    </repositories>
+
     <properties>
         <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
         <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
 
         <scala.compat.version>2.12</scala.compat.version>
         <spark.version>2.4.4</spark.version>
+        <hadoop.version>2.7.3.2.6.1.0-129</hadoop.version>
         <slick.version>3.3.3</slick.version>
         <tminglei.version>0.19.6</tminglei.version>
         <slick-hikaricp.version>3.3.1</slick-hikaricp.version>
@@ -182,11 +190,64 @@
             <artifactId>kafka_${scala.compat.version}</artifactId>
             <version>${kafka.version}</version>
         </dependency>
+
+        <!-- Dependencies for Spark InProcessLauncher -->
         <dependency>
             <groupId>org.apache.spark</groupId>
-            <artifactId>spark-launcher_${scala.compat.version}</artifactId>
+            <artifactId>spark-yarn_${scala.compat.version}</artifactId>
             <version>${spark.version}</version>
+            <exclusions>
+                <exclusion>
+                    <groupId>javax.validation</groupId>
+                    <artifactId>validation-api</artifactId>
+                </exclusion>
+                <exclusion>
+                    <groupId>org.slf4j</groupId>
+                    <artifactId>slf4j-log4j12</artifactId>
+                </exclusion>
+                <exclusion>
+                    <groupId>org.apache.hadoop</groupId>
+                    <artifactId>*</artifactId>
+                </exclusion>
+            </exclusions>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.hadoop</groupId>
+            <artifactId>hadoop-yarn-api</artifactId>
+            <version>${hadoop.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.hadoop</groupId>
+            <artifactId>hadoop-yarn-common</artifactId>
+            <version>${hadoop.version}</version>
         </dependency>
+        <dependency>
+            <groupId>org.apache.hadoop</groupId>
+            <artifactId>hadoop-yarn-server-common</artifactId>
+            <version>${hadoop.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.hadoop</groupId>
+            <artifactId>hadoop-yarn-client</artifactId>
+            <version>${hadoop.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.hadoop</groupId>
+            <artifactId>hadoop-client</artifactId>
+            <version>${hadoop.version}</version>
+            <exclusions>
+                <exclusion>
+                    <groupId>com.google.code.gson</groupId>
+                    <artifactId>gson</artifactId>
+                </exclusion>
+                <exclusion>
+                    <groupId>org.slf4j</groupId>
+                    <artifactId>slf4j-log4j12</artifactId>
+                </exclusion>
+            </exclusions>
+        </dependency>
+        <!-- End Dependencies for Spark InProcessLauncher -->
+
         <dependency>
             <groupId>com.typesafe.play</groupId>
             <artifactId>play-json_${scala.compat.version}</artifactId>

src/main/resources/application.properties

@@ -96,8 +96,6 @@ kafkaSource.properties.sasl.jaas.config=
 spark.submitApi=yarn
 sparkYarnSink.hadoopResourceManagerUrlBase=http://localhost:8088
 sparkYarnSink.userUsedToKillJob=
-sparkYarnSink.hadoopConfDir=/opt/hadoop
-sparkYarnSink.sparkHome=/opt/spark
 sparkYarnSink.master=yarn
 sparkYarnSink.submitTimeout=160000
 sparkYarnSink.filesToDeploy=

src/main/scala/org/apache/spark/launcher/NoBackendConnectionInProcessLauncher.scala

@@ -0,0 +1,49 @@
+
+/*
+ * Copyright 2018 ABSA Group Limited
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.launcher
+
+import org.slf4j.LoggerFactory
+
+class NoBackendConnectionInProcessLauncher extends InProcessLauncher {
+
+  private val logger = LoggerFactory.getLogger(this.getClass)
+  override def startApplication(listeners: SparkAppHandle.Listener*): SparkAppHandle = {
+    import scala.collection.JavaConverters._
+    if (builder.isClientMode(Map[String, String]().asJava)) {
+      logger.warn("It's not recommended to run client-mode applications using InProcessLauncher.")
+    }
+    val main = findSparkSubmit()
+    val server = LauncherServer.getOrCreateServer()
+
+    val handle = new InProcessAppHandle(server)
+    listeners.foreach(handle.addListener)
+
+    // Remove launcher config to make sure that no backend connection is created
+    // See org.apache.spark.launcher.LauncherBackend:connect
+    builder.conf.remove(LauncherProtocol.CONF_LAUNCHER_PORT)
+    builder.conf.remove(LauncherProtocol.CONF_LAUNCHER_SECRET)
+
+    // Set waitAppCompletion to false to ensure fire and forget mode
+    // See org.apache.spark.deploy.yarn.Client:run
+    setConf("spark.yarn.submit.waitAppCompletion", "false")
+
+    val sparkArgs = builder.buildSparkSubmitArgs().asScala.toArray
+    val appName = CommandBuilderUtils.firstNonEmpty(builder.appName, builder.mainClass, "<unknown>")
+    handle.start(appName, main, sparkArgs)
+    handle
+  }
+}

src/main/scala/za/co/absa/hyperdrive/trigger/configuration/application/SparkConfig.scala

@@ -50,9 +50,7 @@ class SparkConfig (
 
 class SparkYarnSinkConfig (
   val submitTimeout: Int,
-  val hadoopConfDir: String,
   val master: String,
-  val sparkHome: String,
   @Name("filesToDeploy")
   filesToDeployInternal: String,
   @Name("additionalConfs")

src/main/scala/za/co/absa/hyperdrive/trigger/configuration/application/SparkConfigNestedClassesValidator.scala

@@ -50,12 +50,8 @@ class SparkConfigNestedClassesValidator extends ConstraintValidator[SparkConfigN
       validateConstraints(Seq(
         Constraint(sparkConfig.yarn.submitTimeout > 0,
           "sparkYarnSink.submitTimeout", "must be > 0"),
-        Constraint(notBlankValidator.isValid(sparkConfig.yarn.hadoopConfDir, context),
-          "sparkYarnSink.hadoopConfDir", notBlankMessage),
         Constraint(notBlankValidator.isValid(sparkConfig.yarn.master, context),
-          "sparkYarnSink.master", notBlankMessage),
-        Constraint(notBlankValidator.isValid(sparkConfig.yarn.sparkHome, context),
-          "sparkYarnSink.sparkHome", notBlankMessage)
+          "sparkYarnSink.master", notBlankMessage)
       ))
     }
   }

src/main/scala/za/co/absa/hyperdrive/trigger/scheduler/executors/spark/SparkYarnClusterServiceImpl.scala

@@ -16,8 +16,7 @@
 
 package za.co.absa.hyperdrive.trigger.scheduler.executors.spark
 
-import org.apache.spark.launcher.{SparkAppHandle, SparkLauncher}
-import org.slf4j.LoggerFactory
+import org.apache.spark.launcher.{InProcessLauncher, NoBackendConnectionInProcessLauncher, SparkAppHandle, SparkLauncher}
 import org.springframework.stereotype.Service
 import za.co.absa.hyperdrive.trigger.configuration.application.SparkConfig
 import za.co.absa.hyperdrive.trigger.models.enums.JobStatuses.{Lost, SubmissionTimeout, Submitting}
@@ -66,23 +65,17 @@ class SparkYarnClusterServiceImpl @Inject()(
   }
 
   private def getSparkLauncher(id: String, jobName: String, jobParameters: SparkInstanceParameters)
-                              (implicit sparkConfig: SparkConfig): SparkLauncher = {
-    import scala.collection.JavaConverters._
+                              (implicit sparkConfig: SparkConfig): InProcessLauncher = {
     val config = sparkConfig.yarn
-    val sparkLauncher = new SparkLauncher(Map(
-      "HADOOP_CONF_DIR" -> config.hadoopConfDir,
-      "SPARK_PRINT_LAUNCH_COMMAND" -> "1"
-    ).asJava)
+    val sparkLauncher = new NoBackendConnectionInProcessLauncher()
       .setMaster(config.master)
       .setDeployMode("cluster")
       .setMainClass(jobParameters.mainClass)
       .setAppResource(jobParameters.jobJar)
-      .setSparkHome(config.sparkHome)
       .setAppName(jobName)
       .setConf("spark.yarn.tags", id)
       .addAppArgs(jobParameters.appArguments.toSeq:_*)
       .addSparkArg("--verbose")
-      .redirectToLog(LoggerFactory.getLogger(s"SparkExecutor.executorJobId=$id").getName)
     config.filesToDeploy.foreach(file => sparkLauncher.addFile(file))
     config.additionalConfs.foreach(conf => sparkLauncher.setConf(conf._1, conf._2))
     jobParameters.additionalJars.foreach(sparkLauncher.addJar)

src/test/resources/application.properties

@@ -41,8 +41,6 @@ kafka-source.properties.security.protocol=PLAINTEXT
 
 spark.submitApi=yarn
 sparkYarnSink.hadoopResourceManagerUrlBase=http://localhost:8088
-sparkYarnSink.hadoopConfDir=/opt/hadoop
-sparkYarnSink.sparkHome=/opt/spark
 sparkYarnSink.master=yarn
 sparkYarnSink.submitTimeout=160000
 

src/test/scala/za/co/absa/hyperdrive/trigger/ApplicationStartPostgresTest.scala

@@ -78,8 +78,6 @@ class ApplicationStartPostgresTest extends FlatSpec with Matchers with SpringInt
     kafkaConfig.properties.getProperty("security.protocol") shouldBe "PLAINTEXT"
     sparkConfig.submitApi shouldBe "yarn"
     sparkConfig.hadoopResourceManagerUrlBase shouldBe "http://localhost:8088"
-    sparkConfig.yarn.hadoopConfDir shouldBe "/opt/hadoop"
-    sparkConfig.yarn.sparkHome shouldBe "/opt/spark"
     sparkConfig.yarn.master shouldBe "yarn"
     sparkConfig.yarn.submitTimeout shouldBe 160000
     sparkConfig.yarn.filesToDeploy shouldBe Seq()

src/test/scala/za/co/absa/hyperdrive/trigger/configuration/application/DefaultTestSparkConfig.scala

@@ -21,9 +21,7 @@ import java.util.Properties
 case class DefaultTestSparkConfig (
   submitApi: String = "yarn",
   submitTimeout: Int = 1000,
-  hadoopConfDir: String = "",
   master: String = "yarn",
-  sparkHome: String = "",
   hadoopResourceManagerUrlBase: String = "",
   filesToDeploy: Seq[String] = Seq(),
   additionalConfs: Map[String, String] = Map(),
@@ -32,7 +30,7 @@ case class DefaultTestSparkConfig (
   clusterId: String = "j-2AXXXXXXGAPLF",
 ) {
   def yarn: SparkConfig =
-    new SparkConfig(submitApi, new SparkYarnSinkConfig(submitTimeout, hadoopConfDir, master, sparkHome,
+    new SparkConfig(submitApi, new SparkYarnSinkConfig(submitTimeout, master,
       filesToDeploy.mkString(","), toProperties(additionalConfs)), null, hadoopResourceManagerUrlBase,
       userUsedToKillJob, sparkSubmitThreadPoolSize
     )