<commit_before>package com.cloudera.hue.livy.server.sessions

import java.lang.ProcessBuilder.Redirect
import java.net.URL

import com.cloudera.hue.livy.Logging
import com.cloudera.hue.livy.server.sessions.Session.SessionFailedToStart

import scala.annotation.tailrec
import scala.concurrent.Future
import scala.io.Source

object ProcessSession extends Logging {
  val LIVY_HOME = System.getenv("LIVY_HOME")
  val LIVY_REPL = LIVY_HOME + "/bin/livy-repl"

  def create(id: String, lang: String): Session = {
    val process = startProcess(id, lang)
    new ProcessSession(id, process)
  }

  // Loop until we've started a process with a valid port.
  private def startProcess(id: String, lang: String): Process = {
    val pb = new ProcessBuilder(LIVY_REPL, lang)

    val callbackUrl = System.getProperty("livy.server.callback-url")
    pb.environment().put("LIVY_CALLBACK_URL", f"$callbackUrl/sessions/$id/callback")
    pb.environment().put("LIVY_PORT", "0")

    pb.redirectOutput(Redirect.INHERIT)
    pb.redirectError(Redirect.INHERIT)

    pb.start()

  }
}

private class ProcessSession(id: String, process: Process) extends WebSession(id) {

  override def stop(): Future[Unit] = {
    super.stop() andThen { case r =>
      // Make sure the process is reaped.
      process.waitFor()

      r
    }
  }
}<commit_msg>[livy] Process sessions should use spark-submit<commit_after>package com.cloudera.hue.livy.server.sessions

import java.lang.ProcessBuilder.Redirect
import java.net.URL

import com.cloudera.hue.livy.{Utils, Logging}
import com.cloudera.hue.livy.server.sessions.Session.SessionFailedToStart

import scala.annotation.tailrec
import scala.concurrent.Future
import scala.io.Source

object ProcessSession extends Logging {
  def create(id: String, lang: String): Session = {
    val process = startProcess(id, lang)
    new ProcessSession(id, process)
  }

  // Loop until we've started a process with a valid port.
  private def startProcess(id: String, lang: String): Process = {
    val pb = new ProcessBuilder(
      "spark-submit",
      "--class", "com.cloudera.hue.livy.repl.Main",
      Utils.jarOfClass(getClass).head,
      lang)

    val callbackUrl = System.getProperty("livy.server.callback-url")
    pb.environment().put("LIVY_CALLBACK_URL", f"$callbackUrl/sessions/$id/callback")
    pb.environment().put("LIVY_PORT", "0")

    pb.redirectOutput(Redirect.INHERIT)
    pb.redirectError(Redirect.INHERIT)

    pb.start()

  }
}

private class ProcessSession(id: String, process: Process) extends WebSession(id) {

  override def stop(): Future[Unit] = {
    super.stop() andThen { case r =>
      // Make sure the process is reaped.
      process.waitFor()

      r
    }
  }
}
{'fraction_non_alphanumeric': 0.11951309479896717, 'fraction_numerical': 0.0007377351530800443, 'mean_word_length': 3.834224598930481, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4829128', 'n_tokens_mistral': 905, 'n_tokens_neox': 886, 'n_words': 195}