Repository: zeppelin Updated Branches: refs/heads/branch-0.8 4660ab073 -> 7ba5e6ffc
ZEPPELIN-3171. Restart of interpreter in note also aborts running interpreter in another note ### What is this PR for? The root cause is that in isolated mode interpreters will share the same scheduler. That means when one interpreter is terminated, all the running jobs under the scheduler of this interpreter will be aborted too. This PR will create one scheduler for each session. So when one session is closed, only the running jobs under this session's scheduler is aborted. ### What type of PR is it? [Bug Fix] ### Todos * [ ] - Task ### What is the Jira issue? * https://issues.apache.org/jira/browse/ZEPPELIN-3171 ### How should this be tested? * Unit test is added. * Also verify it manually. ### Screenshots (if appropriate) ### Questions: * Does the licenses files need update? No * Is there breaking changes for older versions? No * Does this needs documentation? No Author: Jeff Zhang <zjf...@apache.org> Closes #2769 from zjffdu/ZEPPELIN-3171 and squashes the following commits: 3586f45 [Jeff Zhang] ZEPPELIN-3171. Restart of interpreter in note also aborts running interpreter in another note (cherry picked from commit d07c70a6dc32d3d2668198b2e4c10c57602f8ab8) Signed-off-by: Jeff Zhang <zjf...@apache.org> Project: http://git-wip-us.apache.org/repos/asf/zeppelin/repo Commit: http://git-wip-us.apache.org/repos/asf/zeppelin/commit/7ba5e6ff Tree: http://git-wip-us.apache.org/repos/asf/zeppelin/tree/7ba5e6ff Diff: http://git-wip-us.apache.org/repos/asf/zeppelin/diff/7ba5e6ff Branch: refs/heads/branch-0.8 Commit: 7ba5e6ffc6145d0e5426f821878ccc7be8067d7c Parents: 4660ab0 Author: Jeff Zhang <zjf...@apache.org> Authored: Wed Feb 7 14:47:46 2018 +0800 Committer: Jeff Zhang <zjf...@apache.org> Committed: Thu Feb 8 13:57:23 2018 +0800 ---------------------------------------------------------------------- .../apache/zeppelin/interpreter/remote/RemoteInterpreter.java | 7 ++++--- .../zeppelin/interpreter/remote/RemoteInterpreterTest.java | 6 ++++++ 2 files changed, 10 insertions(+), 3 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/zeppelin/blob/7ba5e6ff/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreter.java ---------------------------------------------------------------------- diff --git a/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreter.java b/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreter.java index bda8010..f38d037 100644 --- a/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreter.java +++ b/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreter.java @@ -380,15 +380,16 @@ public class RemoteInterpreter extends Interpreter { }); } - //TODO(zjffdu) Share the Scheduler in the same session or in the same InterpreterGroup ? + @Override public Scheduler getScheduler() { int maxConcurrency = Integer.parseInt( getProperty("zeppelin.interpreter.max.poolsize", ZeppelinConfiguration.ConfVars.ZEPPELIN_INTERPRETER_MAX_POOL_SIZE.getIntValue() + "")); - + // one session own one Scheduler, so that when one session is closed, all the jobs/paragraphs + // running under the scheduler of this session will be aborted. Scheduler s = new RemoteScheduler( - RemoteInterpreter.class.getName() + "-" + sessionId, + RemoteInterpreter.class.getName() + "-" + getInterpreterGroup().getId() + "-" + sessionId, SchedulerFactory.singleton().getExecutor(), sessionId, this, http://git-wip-us.apache.org/repos/asf/zeppelin/blob/7ba5e6ff/zeppelin-zengine/src/test/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreterTest.java ---------------------------------------------------------------------- diff --git a/zeppelin-zengine/src/test/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreterTest.java b/zeppelin-zengine/src/test/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreterTest.java index 7f9978a..04b7a5b 100644 --- a/zeppelin-zengine/src/test/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreterTest.java +++ b/zeppelin-zengine/src/test/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreterTest.java @@ -93,6 +93,8 @@ public class RemoteInterpreterTest { assertTrue(interpreter2 instanceof RemoteInterpreter); RemoteInterpreter remoteInterpreter2 = (RemoteInterpreter) interpreter2; + assertEquals(remoteInterpreter1.getScheduler(), remoteInterpreter2.getScheduler()); + InterpreterContext context1 = new InterpreterContext("noteId", "paragraphId", "repl", "title", "text", AuthenticationInfo.ANONYMOUS, new HashMap<String, Object>(), new GUI(), new GUI(), null, null, new ArrayList<InterpreterContextRunner>(), null); @@ -136,6 +138,8 @@ public class RemoteInterpreterTest { assertTrue(interpreter2 instanceof RemoteInterpreter); RemoteInterpreter remoteInterpreter2 = (RemoteInterpreter) interpreter2; + assertNotEquals(interpreter1.getScheduler(), interpreter2.getScheduler()); + InterpreterContext context1 = new InterpreterContext("noteId", "paragraphId", "repl", "title", "text", AuthenticationInfo.ANONYMOUS, new HashMap<String, Object>(), new GUI(), new GUI(), null, null, new ArrayList<InterpreterContextRunner>(), null); @@ -182,6 +186,8 @@ public class RemoteInterpreterTest { assertTrue(interpreter2 instanceof RemoteInterpreter); RemoteInterpreter remoteInterpreter2 = (RemoteInterpreter) interpreter2; + assertNotEquals(interpreter1.getScheduler(), interpreter2.getScheduler()); + InterpreterContext context1 = new InterpreterContext("noteId", "paragraphId", "repl", "title", "text", AuthenticationInfo.ANONYMOUS, new HashMap<String, Object>(), new GUI(), new GUI(), null, null, new ArrayList<InterpreterContextRunner>(), null);
