You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@toree.apache.org by lr...@apache.org on 2017/08/03 15:52:34 UTC
incubator-toree git commit: [TOREE-415] Cancel Spark jobs when
interrupted
Repository: incubator-toree
Updated Branches:
refs/heads/master 79d02e286 -> 6be462faa
[TOREE-415] Cancel Spark jobs when interrupted
Closes #123
Project: http://git-wip-us.apache.org/repos/asf/incubator-toree/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-toree/commit/6be462fa
Tree: http://git-wip-us.apache.org/repos/asf/incubator-toree/tree/6be462fa
Diff: http://git-wip-us.apache.org/repos/asf/incubator-toree/diff/6be462fa
Branch: refs/heads/master
Commit: 6be462faa13095a51f83f479a468630438e0a437
Parents: 79d02e2
Author: Ryan Blue <bl...@apache.org>
Authored: Fri Jun 2 11:37:31 2017 -0700
Committer: Luciano Resende <lr...@apache.org>
Committed: Thu Jul 20 08:21:49 2017 -0700
----------------------------------------------------------------------
.../interpreter/scala/ScalaInterpreter.scala | 31 +++++++++++++---
.../scala-2.11/scala/ScalaInterpreterSpec.scala | 38 ++++++++++++++++++--
2 files changed, 61 insertions(+), 8 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-toree/blob/6be462fa/scala-interpreter/src/main/scala/org/apache/toree/kernel/interpreter/scala/ScalaInterpreter.scala
----------------------------------------------------------------------
diff --git a/scala-interpreter/src/main/scala/org/apache/toree/kernel/interpreter/scala/ScalaInterpreter.scala b/scala-interpreter/src/main/scala/org/apache/toree/kernel/interpreter/scala/ScalaInterpreter.scala
index 50e8c0a..250d984 100644
--- a/scala-interpreter/src/main/scala/org/apache/toree/kernel/interpreter/scala/ScalaInterpreter.scala
+++ b/scala-interpreter/src/main/scala/org/apache/toree/kernel/interpreter/scala/ScalaInterpreter.scala
@@ -18,7 +18,8 @@
package org.apache.toree.kernel.interpreter.scala
import java.io.ByteArrayOutputStream
-import java.util.concurrent.ExecutionException
+import java.nio.charset.Charset
+import java.util.concurrent.{ExecutionException, TimeoutException, TimeUnit}
import com.typesafe.config.{Config, ConfigFactory}
import jupyter.Displayers
import org.apache.spark.SparkContext
@@ -38,13 +39,16 @@ import scala.tools.nsc.Settings
import scala.tools.nsc.interpreter.{IR, OutputStream}
import scala.tools.nsc.util.ClassPath
import scala.util.matching.Regex
+import scala.concurrent.duration.Duration
class ScalaInterpreter(private val config:Config = ConfigFactory.load) extends Interpreter with ScalaInterpreterSpecific {
import ScalaInterpreter._
ScalaDisplayers.ensureLoaded()
- private var kernel: KernelLike = _
+ protected var _kernel: KernelLike = _
+
+ protected def kernel: KernelLike = _kernel
protected val logger = LoggerFactory.getLogger(this.getClass.getName)
@@ -83,7 +87,7 @@ class ScalaInterpreter(private val config:Config = ConfigFactory.load) extends I
* @return The newly initialized interpreter
*/
override def init(kernel: KernelLike): Interpreter = {
- this.kernel = kernel
+ this._kernel = kernel
val args = interpreterArgs(kernel)
settings = newSettings(args)
settings = appendClassPath(settings)
@@ -154,8 +158,25 @@ class ScalaInterpreter(private val config:Config = ConfigFactory.load) extends I
override def interrupt(): Interpreter = {
require(taskManager != null)
- // Force dumping of current task (begin processing new tasks)
- taskManager.restart()
+ // TODO: use SparkContext.setJobGroup to avoid killing all jobs
+ kernel.sparkContext.cancelAllJobs()
+
+ // give the task 100ms to complete before restarting the task manager
+ import scala.concurrent.ExecutionContext.Implicits.global
+ val finishedFuture = Future {
+ while (taskManager.isExecutingTask) {
+ Thread.sleep(10)
+ }
+ }
+
+ try {
+ Await.result(finishedFuture, Duration(100, TimeUnit.MILLISECONDS))
+ // Await returned, no need to interrupt tasks.
+ } catch {
+ case timeout: TimeoutException =>
+ // Force dumping of current task (begin processing new tasks)
+ taskManager.restart()
+ }
this
}
http://git-wip-us.apache.org/repos/asf/incubator-toree/blob/6be462fa/scala-interpreter/src/test/scala-2.11/scala/ScalaInterpreterSpec.scala
----------------------------------------------------------------------
diff --git a/scala-interpreter/src/test/scala-2.11/scala/ScalaInterpreterSpec.scala b/scala-interpreter/src/test/scala-2.11/scala/ScalaInterpreterSpec.scala
index f95c518..dfe3da6 100644
--- a/scala-interpreter/src/test/scala-2.11/scala/ScalaInterpreterSpec.scala
+++ b/scala-interpreter/src/test/scala-2.11/scala/ScalaInterpreterSpec.scala
@@ -19,15 +19,16 @@ package org.apache.toree.kernel.interpreter.scala
import java.io.{InputStream, OutputStream}
import java.net.{URL, URLClassLoader}
-
+import org.apache.spark.SparkContext
+import org.apache.spark.sql.SparkSession
import org.apache.toree.interpreter.Results.Result
import org.apache.toree.interpreter._
+import org.apache.toree.kernel.api.KernelLike
import org.apache.toree.utils.TaskManager
import org.mockito.Matchers._
import org.mockito.Mockito._
import org.scalatest.mock.MockitoSugar
import org.scalatest.{BeforeAndAfter, FunSpec, Matchers}
-
import scala.concurrent.Future
import scala.tools.nsc.Settings
import scala.tools.nsc.interpreter.{IMain, IR, JPrintWriter}
@@ -41,6 +42,9 @@ class ScalaInterpreterSpec extends FunSpec
private var mockSparkIMain: IMain = _
private var mockTaskManager: TaskManager = _
private var mockSettings: Settings = _
+ private var mockKernel: KernelLike = _
+ private var mockSparkSession: SparkSession = _
+ private var mockSparkContext: SparkContext = _
trait StubbedUpdatePrintStreams extends Interpreter {
override def updatePrintStreams(
@@ -91,6 +95,8 @@ class ScalaInterpreterSpec extends FunSpec
override def newTaskManager(): TaskManager = mockTaskManager
override def newSettings(args: List[String]): Settings = mockSettings
+ override protected def kernel: KernelLike = mockKernel
+
// mocking out these
override protected def reinitializeSymbols(): Unit = {}
override protected def refreshDefinitions(): Unit = {}
@@ -110,6 +116,12 @@ class ScalaInterpreterSpec extends FunSpec
doReturn(mockSettingsClasspath).when(mockSettings).classpath
doNothing().when(mockSettings).embeddedDefaults(any[ClassLoader])
+ mockKernel = mock[KernelLike]
+ mockSparkSession = mock[SparkSession]
+ mockSparkContext = mock[SparkContext]
+ doReturn(mockSparkSession).when(mockKernel).sparkSession
+ doReturn(mockSparkContext).when(mockKernel).sparkContext
+
interpreter = new StubbedStartInterpreter
interpreterNoPrintStreams =
@@ -120,6 +132,9 @@ class ScalaInterpreterSpec extends FunSpec
mockSparkIMain = null
mockTaskManager = null
mockSettings = null
+ mockKernel = null
+ mockSparkSession = null
+ mockSparkContext = null
interpreter = null
}
@@ -193,13 +208,30 @@ class ScalaInterpreterSpec extends FunSpec
}
}
- it("should call restart() on the task manager") {
+ it("should call restart() on the task manager and cancelAllJobs on SparkContext") {
interpreterNoPrintStreams.start()
+ // cancelAllJobs still leaves the task running
+ doReturn(true).when(mockTaskManager).isExecutingTask
+
interpreterNoPrintStreams.interrupt()
+ // restart is called
+ verify(mockSparkContext).cancelAllJobs()
verify(mockTaskManager).restart()
}
+
+ it("should only call cancelAllJobs and not restart if task execution ends") {
+ interpreterNoPrintStreams.start()
+
+ interpreterNoPrintStreams.interrupt()
+
+ // Spark jobs are cancelled
+ verify(mockSparkContext).cancelAllJobs()
+ // The task manager is not executing, so it is not restarted
+ verify(mockTaskManager, atLeastOnce).isExecutingTask
+ verifyNoMoreInteractions(mockSparkContext)
+ }
}
// TODO: Provide testing for the helper functions that return various