DSP-15163 context-per-jvm as subprocess of SJS

SJS used to spawn shell process that spawned context-per-jvm as a bg
process (&). Spawning shell ended and context-per-jvm parent process was
set to init process (1). This commit changes the script so that it does
not fork new process with &.

Author: jtgrabowski

bin/manager_start.sh

@@ -40,5 +40,5 @@ else
   $appdir/spark-job-server.jar $1 $2 $conffile'
 fi
 
-eval $cmd > /dev/null 2>&1 &
+eval $cmd > /dev/null 2>&1
 # exec java -cp $CLASSPATH $GC_OPTS $JAVA_OPTS $LOGGING_OPTS $CONFIG_OVERRIDES $MAIN $1 $2 $conffile 2>&1 &

job-server/src/main/scala/spark/jobserver/AkkaClusterSupervisorActor.scala

@@ -3,20 +3,23 @@ package spark.jobserver
 import java.io.IOException
 import java.nio.file.{Files, Paths}
 import java.nio.charset.Charset
-import java.util.concurrent.TimeUnit
+import java.util.concurrent.{ExecutorService, Executors, TimeUnit}
 
 import akka.actor._
 import akka.cluster.Cluster
 import akka.cluster.ClusterEvent.{InitialStateAsEvents, MemberEvent, MemberUp}
 import akka.util.Timeout
+import com.google.common.util.concurrent.ThreadFactoryBuilder
 import com.typesafe.config.{Config, ConfigFactory, ConfigRenderOptions}
 import spark.jobserver.util.SparkJobUtils
+
 import scala.collection.mutable
 import scala.util.{Failure, Success, Try}
 import scala.sys.process._
-
 import spark.jobserver.common.akka.InstrumentedActor
 
+import scala.collection.concurrent.TrieMap
+
 /**
  * The AkkaClusterSupervisorActor launches Spark Contexts as external processes
  * that connect back with the master node via Akka Cluster.
@@ -51,8 +54,10 @@ class AkkaClusterSupervisorActor(daoActor: ActorRef) extends InstrumentedActor {
   //TODO: try to pass this state to the jobManager at start instead of having to track
   //extra state.  What happens if the WebApi process dies before the forked process
   //starts up?  Then it never gets initialized, and this state disappears.
-  private val contextInitInfos = mutable.HashMap.empty[String, (Boolean, ActorRef => Unit, Throwable => Unit)]
-
+  private val contextInitInfos = TrieMap.empty[String, (Boolean, ActorRef => Unit, Throwable => Unit)]
+  private val contextInitExecutorService = Executors.newCachedThreadPool(
+    new ThreadFactoryBuilder().setDaemon(true).setNameFormat("job-server-context-init-thread -% d").build
+  )
   // actor name -> (JobManagerActor ref, ResultActor ref)
   private val contexts = mutable.HashMap.empty[String, (ActorRef, ActorRef)]
 
@@ -214,26 +219,31 @@ class AkkaClusterSupervisorActor(daoActor: ActorRef) extends InstrumentedActor {
       cmdString = cmdString + s" ${contextConfig.getString(SparkJobUtils.SPARK_PROXY_USER_PARAM)}"
     }
 
-    val pb = Process(cmdString)
-    val pio = new ProcessIO(_ => (),
-                        stdout => scala.io.Source.fromInputStream(stdout)
-                          .getLines.foreach(println),
-                        stderr => scala.io.Source.fromInputStream(stderr).getLines().foreach(println))
-    logger.info("Starting to execute sub process {}", pb)
-    val processStart = Try {
-      val process = pb.run(pio)
-      val exitVal = process.exitValue()
-      if (exitVal != 0) {
-        throw new IOException("Failed to launch context process, got exit code " + exitVal)
-      }
-    }
-
-    if (processStart.isSuccess) {
-      contextInitInfos(contextActorName) = (isAdHoc, successFunc, failureFunc)
-    } else {
-      failureFunc(processStart.failed.get)
-    }
+    contextInitInfos(contextActorName) = (isAdHoc, successFunc, failureFunc)
+
+    contextInitExecutorService.submit(new Runnable {
+      override def run(): Unit = {
+        val pb = Process(cmdString)
+        val pio = new ProcessIO(_ => (),
+          stdout => scala.io.Source.fromInputStream(stdout)
+            .getLines.foreach(println),
+          stderr => scala.io.Source.fromInputStream(stderr).getLines().foreach(println))
+
+        logger.info("Starting to execute sub process {}", pb)
+        val processStart = Try {
+          val process = pb.run(pio)
+          val exitVal = process.exitValue()
+          if (exitVal != 0) {
+            throw new IOException("Failed to launch context process, got exit code " + exitVal)
+          }
+        }
 
+        if (processStart.isFailure) {
+          failureFunc(processStart.failed.get)
+          contextInitInfos.remove(contextActorName)
+        }
+      }
+    })
   }
 
   private def createContextDir(name: String,