Repository: zeppelin Updated Branches: refs/heads/master 4b436ca22 -> e9dedab46
ZEPPELIN-3506. DepInterpreter is broken ### What is this PR for? The bug is due to getInterpreterInTheSameSessionByClassName doesn't find the correct DepInterpreter. This PR fix this issue. The unit test fails due to classpath issue, will enable it later. ### What type of PR is it? [Bug Fix] ### Todos * [ ] - Task ### What is the Jira issue? * https://issues.apache.org/jira/browse/ZEPPELIN-3506 ### How should this be tested? * CI pass and manually tested ### 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 #2988 from zjffdu/ZEPPELIN-3506 and squashes the following commits: dd77d5c28 [Jeff Zhang] ZEPPELIN-3506. DepInterpreter is broken Project: http://git-wip-us.apache.org/repos/asf/zeppelin/repo Commit: http://git-wip-us.apache.org/repos/asf/zeppelin/commit/e9dedab4 Tree: http://git-wip-us.apache.org/repos/asf/zeppelin/tree/e9dedab4 Diff: http://git-wip-us.apache.org/repos/asf/zeppelin/diff/e9dedab4 Branch: refs/heads/master Commit: e9dedab46df9dfe3ff6902e453db92cf0e712e82 Parents: 4b436ca Author: Jeff Zhang <zjf...@apache.org> Authored: Mon May 28 10:19:42 2018 +0800 Committer: Jeff Zhang <zjf...@apache.org> Committed: Mon May 28 12:49:57 2018 +0800 ---------------------------------------------------------------------- .../spark/AbstractSparkInterpreter.java | 10 +++++ .../zeppelin/spark/NewSparkInterpreter.java | 3 +- .../zeppelin/spark/OldSparkInterpreter.java | 3 +- .../apache/zeppelin/spark/SparkInterpreter.java | 1 + .../zeppelin/spark/NewSparkInterpreterTest.java | 39 +++++++++++++++++++- 5 files changed, 53 insertions(+), 3 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/zeppelin/blob/e9dedab4/spark/interpreter/src/main/java/org/apache/zeppelin/spark/AbstractSparkInterpreter.java ---------------------------------------------------------------------- diff --git a/spark/interpreter/src/main/java/org/apache/zeppelin/spark/AbstractSparkInterpreter.java b/spark/interpreter/src/main/java/org/apache/zeppelin/spark/AbstractSparkInterpreter.java index 9968dc6..aa1343a 100644 --- a/spark/interpreter/src/main/java/org/apache/zeppelin/spark/AbstractSparkInterpreter.java +++ b/spark/interpreter/src/main/java/org/apache/zeppelin/spark/AbstractSparkInterpreter.java @@ -31,6 +31,8 @@ import java.util.Properties; */ public abstract class AbstractSparkInterpreter extends Interpreter { + private SparkInterpreter parentSparkInterpreter; + public AbstractSparkInterpreter(Properties properties) { super(properties); } @@ -54,4 +56,12 @@ public abstract class AbstractSparkInterpreter extends Interpreter { public abstract String getSparkUIUrl(); public abstract boolean isUnsupportedSparkVersion(); + + public void setParentSparkInterpreter(SparkInterpreter parentSparkInterpreter) { + this.parentSparkInterpreter = parentSparkInterpreter; + } + + public SparkInterpreter getParentSparkInterpreter() { + return parentSparkInterpreter; + } } http://git-wip-us.apache.org/repos/asf/zeppelin/blob/e9dedab4/spark/interpreter/src/main/java/org/apache/zeppelin/spark/NewSparkInterpreter.java ---------------------------------------------------------------------- diff --git a/spark/interpreter/src/main/java/org/apache/zeppelin/spark/NewSparkInterpreter.java b/spark/interpreter/src/main/java/org/apache/zeppelin/spark/NewSparkInterpreter.java index 9b629f9..b2b715b 100644 --- a/spark/interpreter/src/main/java/org/apache/zeppelin/spark/NewSparkInterpreter.java +++ b/spark/interpreter/src/main/java/org/apache/zeppelin/spark/NewSparkInterpreter.java @@ -244,7 +244,8 @@ public class NewSparkInterpreter extends AbstractSparkInterpreter { } private DepInterpreter getDepInterpreter() { - Interpreter p = getInterpreterInTheSameSessionByClassName(DepInterpreter.class.getName()); + Interpreter p = getParentSparkInterpreter() + .getInterpreterInTheSameSessionByClassName(DepInterpreter.class.getName()); if (p == null) { return null; } http://git-wip-us.apache.org/repos/asf/zeppelin/blob/e9dedab4/spark/interpreter/src/main/java/org/apache/zeppelin/spark/OldSparkInterpreter.java ---------------------------------------------------------------------- diff --git a/spark/interpreter/src/main/java/org/apache/zeppelin/spark/OldSparkInterpreter.java b/spark/interpreter/src/main/java/org/apache/zeppelin/spark/OldSparkInterpreter.java index 1f59d18..0dfe3cb 100644 --- a/spark/interpreter/src/main/java/org/apache/zeppelin/spark/OldSparkInterpreter.java +++ b/spark/interpreter/src/main/java/org/apache/zeppelin/spark/OldSparkInterpreter.java @@ -281,7 +281,8 @@ public class OldSparkInterpreter extends AbstractSparkInterpreter { } private DepInterpreter getDepInterpreter() { - Interpreter p = getInterpreterInTheSameSessionByClassName(DepInterpreter.class.getName()); + Interpreter p = getParentSparkInterpreter() + .getInterpreterInTheSameSessionByClassName(DepInterpreter.class.getName()); if (p == null) { return null; } http://git-wip-us.apache.org/repos/asf/zeppelin/blob/e9dedab4/spark/interpreter/src/main/java/org/apache/zeppelin/spark/SparkInterpreter.java ---------------------------------------------------------------------- diff --git a/spark/interpreter/src/main/java/org/apache/zeppelin/spark/SparkInterpreter.java b/spark/interpreter/src/main/java/org/apache/zeppelin/spark/SparkInterpreter.java index d9be573..7df1bc9 100644 --- a/spark/interpreter/src/main/java/org/apache/zeppelin/spark/SparkInterpreter.java +++ b/spark/interpreter/src/main/java/org/apache/zeppelin/spark/SparkInterpreter.java @@ -50,6 +50,7 @@ public class SparkInterpreter extends AbstractSparkInterpreter { } else { delegation = new OldSparkInterpreter(properties); } + delegation.setParentSparkInterpreter(this); } @Override http://git-wip-us.apache.org/repos/asf/zeppelin/blob/e9dedab4/spark/interpreter/src/test/java/org/apache/zeppelin/spark/NewSparkInterpreterTest.java ---------------------------------------------------------------------- diff --git a/spark/interpreter/src/test/java/org/apache/zeppelin/spark/NewSparkInterpreterTest.java b/spark/interpreter/src/test/java/org/apache/zeppelin/spark/NewSparkInterpreterTest.java index 3d22af3..f6cb9a9 100644 --- a/spark/interpreter/src/test/java/org/apache/zeppelin/spark/NewSparkInterpreterTest.java +++ b/spark/interpreter/src/test/java/org/apache/zeppelin/spark/NewSparkInterpreterTest.java @@ -17,11 +17,13 @@ package org.apache.zeppelin.spark; +import com.google.common.io.Files; import org.apache.zeppelin.display.AngularObjectRegistry; import org.apache.zeppelin.display.GUI; import org.apache.zeppelin.display.ui.CheckBox; import org.apache.zeppelin.display.ui.Select; import org.apache.zeppelin.display.ui.TextBox; +import org.apache.zeppelin.interpreter.Interpreter; import org.apache.zeppelin.interpreter.InterpreterContext; import org.apache.zeppelin.interpreter.InterpreterException; import org.apache.zeppelin.interpreter.InterpreterGroup; @@ -30,10 +32,10 @@ import org.apache.zeppelin.interpreter.InterpreterOutputListener; import org.apache.zeppelin.interpreter.InterpreterResult; import org.apache.zeppelin.interpreter.InterpreterResultMessageOutput; import org.apache.zeppelin.interpreter.remote.RemoteEventClient; -import org.apache.zeppelin.interpreter.remote.RemoteEventClientWrapper; import org.apache.zeppelin.interpreter.thrift.InterpreterCompletion; import org.apache.zeppelin.user.AuthenticationInfo; import org.junit.After; +import org.junit.Ignore; import org.junit.Test; import java.io.File; @@ -42,6 +44,7 @@ import java.io.IOException; import java.net.URL; import java.nio.channels.Channels; import java.nio.channels.ReadableByteChannel; +import java.util.ArrayList; import java.util.HashMap; import java.util.List; import java.util.Map; @@ -57,6 +60,7 @@ import static org.mockito.Mockito.verify; public class NewSparkInterpreterTest { private SparkInterpreter interpreter; + private DepInterpreter depInterpreter; // catch the streaming output in onAppend private volatile String output = ""; @@ -351,11 +355,44 @@ public class NewSparkInterpreterTest { assertEquals(InterpreterResult.Code.SUCCESS, result.code()); } + //TODO(zjffdu) This unit test will fail due to classpath issue, should enable it after the classpath issue is fixed. + @Ignore + public void testDepInterpreter() throws InterpreterException { + Properties properties = new Properties(); + properties.setProperty("spark.master", "local"); + properties.setProperty("spark.app.name", "test"); + properties.setProperty("zeppelin.spark.maxResult", "100"); + properties.setProperty("zeppelin.spark.test", "true"); + properties.setProperty("zeppelin.spark.useNew", "true"); + properties.setProperty("zeppelin.dep.localrepo", Files.createTempDir().getAbsolutePath()); + + InterpreterGroup intpGroup = new InterpreterGroup(); + interpreter = new SparkInterpreter(properties); + depInterpreter = new DepInterpreter(properties); + interpreter.setInterpreterGroup(intpGroup); + depInterpreter.setInterpreterGroup(intpGroup); + intpGroup.put("session_1", new ArrayList<Interpreter>()); + intpGroup.get("session_1").add(interpreter); + intpGroup.get("session_1").add(depInterpreter); + + depInterpreter.open(); + InterpreterResult result = + depInterpreter.interpret("z.load(\"com.databricks:spark-avro_2.11:3.2.0\")", getInterpreterContext()); + assertEquals(InterpreterResult.Code.SUCCESS, result.code()); + + interpreter.open(); + result = interpreter.interpret("import com.databricks.spark.avro._", getInterpreterContext()); + assertEquals(InterpreterResult.Code.SUCCESS, result.code()); + } + @After public void tearDown() throws InterpreterException { if (this.interpreter != null) { this.interpreter.close(); } + if (this.depInterpreter != null) { + this.depInterpreter.close(); + } } private InterpreterContext getInterpreterContext() {
