) tezDAGStatsMapField.get(stat);
+ for (TezDAGStats dagStats : tezDAGStatsMap.values()) {
+ LOGGER.debug("Tez JobId:" + dagStats.getJobId());
+ jobIds.add(dagStats.getJobId());
+ }
+ return jobIds;
+ } catch (Exception e) {
+ LOGGER.error("Can not extract jobIds from TezPigScriptStats", e);
+ throw new RuntimeException("Can not extract jobIds from TezPigScriptStats", e);
+ }
+ }
+}
diff --git a/pig/src/main/resources/interpreter-setting.json b/pig/src/main/resources/interpreter-setting.json
new file mode 100644
index 00000000000..27918ede1bf
--- /dev/null
+++ b/pig/src/main/resources/interpreter-setting.json
@@ -0,0 +1,46 @@
+[
+ {
+ "group": "pig",
+ "name": "script",
+ "className": "org.apache.zeppelin.pig.PigInterpreter",
+ "properties": {
+ "zeppelin.pig.execType": {
+ "envName": null,
+ "propertyName": "zeppelin.pig.execType",
+ "defaultValue": "mapreduce",
+ "description": "local | mapreduce | tez"
+ },
+ "zeppelin.pig.includeJobStats": {
+ "envName": null,
+ "propertyName": "zeppelin.pig.includeJobStats",
+ "defaultValue": "false",
+ "description": "flag to include job stats in output"
+ }
+ },
+ "editor": {
+ "language": "pig"
+ }
+ },
+ {
+ "group": "pig",
+ "name": "query",
+ "className": "org.apache.zeppelin.pig.PigQueryInterpreter",
+ "properties": {
+ "zeppelin.pig.execType": {
+ "envName": null,
+ "propertyName": "zeppelin.pig.execType",
+ "defaultValue": "mapreduce",
+ "description": "local | mapreduce | tez"
+ },
+ "zeppelin.pig.maxResult": {
+ "envName": null,
+ "propertyName": "zeppelin.pig.maxResult",
+ "defaultValue": "1000",
+ "description": "max row number for %pig.query"
+ }
+ },
+ "editor": {
+ "language": "pig"
+ }
+ }
+]
diff --git a/pig/src/test/java/org/apache/zeppelin/pig/PigInterpreterTest.java b/pig/src/test/java/org/apache/zeppelin/pig/PigInterpreterTest.java
new file mode 100644
index 00000000000..3d062d61579
--- /dev/null
+++ b/pig/src/test/java/org/apache/zeppelin/pig/PigInterpreterTest.java
@@ -0,0 +1,155 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+
+package org.apache.zeppelin.pig;
+
+import org.apache.commons.io.IOUtils;
+import org.apache.zeppelin.interpreter.InterpreterContext;
+import org.apache.zeppelin.interpreter.InterpreterResult;
+import org.apache.zeppelin.interpreter.InterpreterResult.Code;
+import org.apache.zeppelin.interpreter.InterpreterResult.Type;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.io.File;
+import java.io.FileWriter;
+import java.io.IOException;
+import java.util.Properties;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+public class PigInterpreterTest {
+
+ private PigInterpreter pigInterpreter;
+ private InterpreterContext context;
+
+ @Before
+ public void setUp() {
+ Properties properties = new Properties();
+ properties.put("zeppelin.pig.execType", "local");
+ pigInterpreter = new PigInterpreter(properties);
+ pigInterpreter.open();
+ context = new InterpreterContext(null, "paragraph_id", null, null, null, null, null, null, null,
+ null, null);
+ }
+
+ @After
+ public void tearDown() {
+ pigInterpreter.close();
+ }
+
+ @Test
+ public void testBasics() throws IOException {
+ String content = "1\tandy\n"
+ + "2\tpeter\n";
+ File tmpFile = File.createTempFile("zeppelin", "test");
+ FileWriter writer = new FileWriter(tmpFile);
+ IOUtils.write(content, writer);
+ writer.close();
+
+ // simple pig script using dump
+ String pigscript = "a = load '" + tmpFile.getAbsolutePath() + "';"
+ + "dump a;";
+ InterpreterResult result = pigInterpreter.interpret(pigscript, context);
+ assertEquals(Type.TEXT, result.type());
+ assertEquals(Code.SUCCESS, result.code());
+ assertTrue(result.message().contains("(1,andy)\n(2,peter)"));
+
+ // describe
+ pigscript = "a = load '" + tmpFile.getAbsolutePath() + "' as (id: int, name: bytearray);"
+ + "describe a;";
+ result = pigInterpreter.interpret(pigscript, context);
+ assertEquals(Type.TEXT, result.type());
+ assertEquals(Code.SUCCESS, result.code());
+ assertTrue(result.message().contains("a: {id: int,name: bytearray}"));
+
+ // syntax error (compilation error)
+ pigscript = "a = loa '" + tmpFile.getAbsolutePath() + "';"
+ + "describe a;";
+ result = pigInterpreter.interpret(pigscript, context);
+ assertEquals(Type.TEXT, result.type());
+ assertEquals(Code.ERROR, result.code());
+ assertTrue(result.message().contains("Syntax error, unexpected symbol at or near 'a'"));
+
+ // execution error
+ pigscript = "a = load 'invalid_path';"
+ + "dump a;";
+ result = pigInterpreter.interpret(pigscript, context);
+ assertEquals(Type.TEXT, result.type());
+ assertEquals(Code.ERROR, result.code());
+ assertTrue(result.message().contains("Input path does not exist"));
+ }
+
+
+ @Test
+ public void testIncludeJobStats() throws IOException {
+ Properties properties = new Properties();
+ properties.put("zeppelin.pig.execType", "local");
+ properties.put("zeppelin.pig.includeJobStats", "true");
+ pigInterpreter = new PigInterpreter(properties);
+ pigInterpreter.open();
+
+ String content = "1\tandy\n"
+ + "2\tpeter\n";
+ File tmpFile = File.createTempFile("zeppelin", "test");
+ FileWriter writer = new FileWriter(tmpFile);
+ IOUtils.write(content, writer);
+ writer.close();
+
+ // simple pig script using dump
+ String pigscript = "a = load '" + tmpFile.getAbsolutePath() + "';"
+ + "dump a;";
+ InterpreterResult result = pigInterpreter.interpret(pigscript, context);
+ assertEquals(Type.TEXT, result.type());
+ assertEquals(Code.SUCCESS, result.code());
+ assertTrue(result.message().contains("Counters:"));
+ assertTrue(result.message().contains("(1,andy)\n(2,peter)"));
+
+ // describe
+ pigscript = "a = load '" + tmpFile.getAbsolutePath() + "' as (id: int, name: bytearray);"
+ + "describe a;";
+ result = pigInterpreter.interpret(pigscript, context);
+ assertEquals(Type.TEXT, result.type());
+ assertEquals(Code.SUCCESS, result.code());
+ // no job is launched, so no jobStats
+ assertTrue(!result.message().contains("Counters:"));
+ assertTrue(result.message().contains("a: {id: int,name: bytearray}"));
+
+ // syntax error (compilation error)
+ pigscript = "a = loa '" + tmpFile.getAbsolutePath() + "';"
+ + "describe a;";
+ result = pigInterpreter.interpret(pigscript, context);
+ assertEquals(Type.TEXT, result.type());
+ assertEquals(Code.ERROR, result.code());
+ // no job is launched, so no jobStats
+ assertTrue(!result.message().contains("Counters:"));
+ assertTrue(result.message().contains("Syntax error, unexpected symbol at or near 'a'"));
+
+ // execution error
+ pigscript = "a = load 'invalid_path';"
+ + "dump a;";
+ result = pigInterpreter.interpret(pigscript, context);
+ assertEquals(Type.TEXT, result.type());
+ assertEquals(Code.ERROR, result.code());
+ assertTrue(result.message().contains("Counters:"));
+ assertTrue(result.message().contains("Input path does not exist"));
+ }
+}
diff --git a/pig/src/test/java/org/apache/zeppelin/pig/PigQueryInterpreterTest.java b/pig/src/test/java/org/apache/zeppelin/pig/PigQueryInterpreterTest.java
new file mode 100644
index 00000000000..00ece440542
--- /dev/null
+++ b/pig/src/test/java/org/apache/zeppelin/pig/PigQueryInterpreterTest.java
@@ -0,0 +1,153 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.zeppelin.pig;
+
+import org.apache.commons.io.IOUtils;
+import org.apache.zeppelin.interpreter.Interpreter;
+import org.apache.zeppelin.interpreter.InterpreterContext;
+import org.apache.zeppelin.interpreter.InterpreterGroup;
+import org.apache.zeppelin.interpreter.InterpreterResult;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import java.io.File;
+import java.io.FileWriter;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Properties;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+/**
+ *
+ */
+public class PigQueryInterpreterTest {
+
+ private PigInterpreter pigInterpreter;
+ private PigQueryInterpreter pigQueryInterpreter;
+ private InterpreterContext context;
+
+ @Before
+ public void setUp() {
+ Properties properties = new Properties();
+ properties.put("zeppelin.pig.execType", "local");
+ properties.put("zeppelin.pig.maxResult", "20");
+
+ pigInterpreter = new PigInterpreter(properties);
+ pigQueryInterpreter = new PigQueryInterpreter(properties);
+ List interpreters = new ArrayList();
+ interpreters.add(pigInterpreter);
+ interpreters.add(pigQueryInterpreter);
+ InterpreterGroup group = new InterpreterGroup();
+ group.put("note_id", interpreters);
+ pigInterpreter.setInterpreterGroup(group);
+ pigQueryInterpreter.setInterpreterGroup(group);
+ pigInterpreter.open();
+ pigQueryInterpreter.open();
+
+ context = new InterpreterContext(null, "paragraph_id", null, null, null, null, null, null, null,
+ null, null);
+ }
+
+ @After
+ public void tearDown() {
+ pigInterpreter.close();
+ pigQueryInterpreter.close();
+ }
+
+ @Test
+ public void testBasics() throws IOException {
+ String content = "andy\tmale\t10\n"
+ + "peter\tmale\t20\n"
+ + "amy\tfemale\t14\n";
+ File tmpFile = File.createTempFile("zeppelin", "test");
+ FileWriter writer = new FileWriter(tmpFile);
+ IOUtils.write(content, writer);
+ writer.close();
+
+ // run script in PigInterpreter
+ String pigscript = "a = load '" + tmpFile.getAbsolutePath() + "' as (name, gender, age);\n"
+ + "a2 = load 'invalid_path' as (name, gender, age);\n"
+ + "dump a;";
+ InterpreterResult result = pigInterpreter.interpret(pigscript, context);
+ assertEquals(InterpreterResult.Type.TEXT, result.type());
+ assertEquals(InterpreterResult.Code.SUCCESS, result.code());
+ assertTrue(result.message().contains("(andy,male,10)\n(peter,male,20)\n(amy,female,14)"));
+
+ // run single line query in PigQueryInterpreter
+ String query = "foreach a generate name, age;";
+ result = pigQueryInterpreter.interpret(query, context);
+ assertEquals(InterpreterResult.Type.TABLE, result.type());
+ assertEquals(InterpreterResult.Code.SUCCESS, result.code());
+ assertEquals("name\tage\nandy\t10\npeter\t20\namy\t14\n", result.message());
+
+ // run multiple line query in PigQueryInterpreter
+ query = "b = group a by gender;\nforeach b generate group as gender, COUNT($1) as count;";
+ result = pigQueryInterpreter.interpret(query, context);
+ assertEquals(InterpreterResult.Type.TABLE, result.type());
+ assertEquals(InterpreterResult.Code.SUCCESS, result.code());
+ assertEquals("gender\tcount\nmale\t2\nfemale\t1\n", result.message());
+
+ // syntax error in PigQueryInterpereter
+ query = "b = group a by invalid_column;\nforeach b generate group as gender, COUNT($1) as count;";
+ result = pigQueryInterpreter.interpret(query, context);
+ assertEquals(InterpreterResult.Type.TEXT, result.type());
+ assertEquals(InterpreterResult.Code.ERROR, result.code());
+ assertTrue(result.message().contains("Projected field [invalid_column] does not exist in schema"));
+
+ // execution error in PigQueryInterpreter
+ query = "foreach a2 generate name, age;";
+ result = pigQueryInterpreter.interpret(query, context);
+ assertEquals(InterpreterResult.Type.TEXT, result.type());
+ assertEquals(InterpreterResult.Code.ERROR, result.code());
+ assertTrue(result.message().contains("Input path does not exist"));
+ }
+
+ @Test
+ public void testMaxResult() throws IOException {
+ StringBuilder content = new StringBuilder();
+ for (int i=0;i<30;++i) {
+ content.append(i + "\tname_" + i + "\n");
+ }
+ File tmpFile = File.createTempFile("zeppelin", "test");
+ FileWriter writer = new FileWriter(tmpFile);
+ IOUtils.write(content, writer);
+ writer.close();
+
+ // run script in PigInterpreter
+ String pigscript = "a = load '" + tmpFile.getAbsolutePath() + "' as (id, name);";
+ InterpreterResult result = pigInterpreter.interpret(pigscript, context);
+ assertEquals(InterpreterResult.Type.TEXT, result.type());
+ assertEquals(InterpreterResult.Code.SUCCESS, result.code());
+ // empty output
+ assertTrue(result.message().isEmpty());
+
+ // run single line query in PigQueryInterpreter
+ String query = "foreach a generate id;";
+ result = pigQueryInterpreter.interpret(query, context);
+ assertEquals(InterpreterResult.Type.TABLE, result.type());
+ assertEquals(InterpreterResult.Code.SUCCESS, result.code());
+ assertTrue(result.message().contains("id\n0\n1\n2"));
+ assertTrue(result.message().contains("Results are limited by 20"));
+ }
+}
diff --git a/pig/src/test/resources/log4j.properties b/pig/src/test/resources/log4j.properties
new file mode 100644
index 00000000000..8daee59d60d
--- /dev/null
+++ b/pig/src/test/resources/log4j.properties
@@ -0,0 +1,22 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+log4j.rootLogger = INFO, stdout
+
+log4j.appender.stdout = org.apache.log4j.ConsoleAppender
+log4j.appender.stdout.layout = org.apache.log4j.PatternLayout
+log4j.appender.stdout.layout.ConversionPattern=%5p [%d] ({%t} %F[%M]:%L) - %m%n
diff --git a/pom.xml b/pom.xml
index 03b226341e1..558ce0624a1 100644
--- a/pom.xml
+++ b/pom.xml
@@ -62,6 +62,7 @@
shell
livy
hbase
+ pig
postgresql
jdbc
file
diff --git a/spark-dependencies/pom.xml b/spark-dependencies/pom.xml
index f320680072b..04b6983fb03 100644
--- a/spark-dependencies/pom.xml
+++ b/spark-dependencies/pom.xml
@@ -525,9 +525,9 @@
true
- 2.0.0
+ 2.0.1
2.5.0
- 0.10.1
+ 0.10.3
2.11.8
diff --git a/spark/pom.xml b/spark/pom.xml
index 66d93c42ee6..efb7452647c 100644
--- a/spark/pom.xml
+++ b/spark/pom.xml
@@ -37,7 +37,7 @@
1.8.2
1.10.19
1.6.4
- 2.0.0
+ 2.0.1
@@ -519,9 +519,9 @@
true
- 2.0.0
+ 2.0.1
2.5.0
- 0.10.1
+ 0.10.3
2.11.8
diff --git a/spark/src/main/java/org/apache/zeppelin/spark/SparkInterpreter.java b/spark/src/main/java/org/apache/zeppelin/spark/SparkInterpreter.java
index 878c0106b06..0812c761038 100644
--- a/spark/src/main/java/org/apache/zeppelin/spark/SparkInterpreter.java
+++ b/spark/src/main/java/org/apache/zeppelin/spark/SparkInterpreter.java
@@ -49,6 +49,7 @@
import org.apache.zeppelin.interpreter.Interpreter;
import org.apache.zeppelin.interpreter.InterpreterContext;
import org.apache.zeppelin.interpreter.InterpreterException;
+import org.apache.zeppelin.interpreter.InterpreterHookRegistry;
import org.apache.zeppelin.interpreter.InterpreterProperty;
import org.apache.zeppelin.interpreter.InterpreterResult;
import org.apache.zeppelin.interpreter.InterpreterResult.Code;
@@ -101,6 +102,7 @@ public class SparkInterpreter extends Interpreter {
private SparkConf conf;
private static SparkContext sc;
private static SQLContext sqlc;
+ private static InterpreterHookRegistry hooks;
private static SparkEnv env;
private static Object sparkSession; // spark 2.x
private static JobProgressListener sparkListener;
@@ -479,7 +481,7 @@ private void setupConfForPySpark(SparkConf conf) {
//Only one of py4j-0.9-src.zip and py4j-0.8.2.1-src.zip should exist
String[] pythonLibs = new String[]{"pyspark.zip", "py4j-0.9-src.zip", "py4j-0.8.2.1-src.zip",
- "py4j-0.10.1-src.zip"};
+ "py4j-0.10.1-src.zip", "py4j-0.10.3-src.zip"};
ArrayList pythonLibUris = new ArrayList<>();
for (String lib : pythonLibs) {
File libFile = new File(pysparkPath, lib);
@@ -813,8 +815,10 @@ public void open() {
sqlc = getSQLContext();
dep = getDependencyResolver();
+
+ hooks = getInterpreterGroup().getInterpreterHookRegistry();
- z = new ZeppelinContext(sc, sqlc, null, dep,
+ z = new ZeppelinContext(sc, sqlc, null, dep, hooks,
Integer.parseInt(getProperty("zeppelin.spark.maxResult")));
interpret("@transient val _binder = new java.util.HashMap[String, Object]()");
diff --git a/spark/src/main/java/org/apache/zeppelin/spark/ZeppelinContext.java b/spark/src/main/java/org/apache/zeppelin/spark/ZeppelinContext.java
index 7465756e14f..92b50d0a3b8 100644
--- a/spark/src/main/java/org/apache/zeppelin/spark/ZeppelinContext.java
+++ b/spark/src/main/java/org/apache/zeppelin/spark/ZeppelinContext.java
@@ -28,11 +28,14 @@
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
+import java.util.Map;
+import java.util.HashMap;
import org.apache.spark.SparkContext;
import org.apache.spark.sql.SQLContext;
import org.apache.spark.sql.catalyst.expressions.Attribute;
import org.apache.zeppelin.annotation.ZeppelinApi;
+import org.apache.zeppelin.annotation.Experimental;
import org.apache.zeppelin.display.AngularObject;
import org.apache.zeppelin.display.AngularObjectRegistry;
import org.apache.zeppelin.display.AngularObjectWatcher;
@@ -41,6 +44,7 @@
import org.apache.zeppelin.interpreter.InterpreterContext;
import org.apache.zeppelin.interpreter.InterpreterContextRunner;
import org.apache.zeppelin.interpreter.InterpreterException;
+import org.apache.zeppelin.interpreter.InterpreterHookRegistry;
import org.apache.zeppelin.spark.dep.SparkDependencyResolver;
import org.apache.zeppelin.resource.Resource;
import org.apache.zeppelin.resource.ResourcePool;
@@ -53,19 +57,33 @@
* Spark context for zeppelin.
*/
public class ZeppelinContext {
+ // Map interpreter class name (to be used by hook registry) from
+ // given replName in parapgraph
+ private static final Map interpreterClassMap;
+ static {
+ interpreterClassMap = new HashMap();
+ interpreterClassMap.put("spark", "org.apache.zeppelin.spark.SparkInterpreter");
+ interpreterClassMap.put("sql", "org.apache.zeppelin.spark.SparkSqlInterpreter");
+ interpreterClassMap.put("dep", "org.apache.zeppelin.spark.DepInterpreter");
+ interpreterClassMap.put("pyspark", "org.apache.zeppelin.spark.PySparkInterpreter");
+ }
+
private SparkDependencyResolver dep;
private InterpreterContext interpreterContext;
private int maxResult;
private List supportedClasses;
-
+ private InterpreterHookRegistry hooks;
+
public ZeppelinContext(SparkContext sc, SQLContext sql,
InterpreterContext interpreterContext,
SparkDependencyResolver dep,
+ InterpreterHookRegistry hooks,
int maxResult) {
this.sc = sc;
this.sqlContext = sql;
this.interpreterContext = interpreterContext;
this.dep = dep;
+ this.hooks = hooks;
this.maxResult = maxResult;
this.supportedClasses = new ArrayList<>();
try {
@@ -697,6 +715,90 @@ private void angularUnbind(String name, String noteId) {
registry.remove(name, noteId, null);
}
+ /**
+ * Get the interpreter class name from name entered in paragraph
+ * @param replName if replName is a valid className, return that instead.
+ */
+ public String getClassNameFromReplName(String replName) {
+ for (String name : interpreterClassMap.values()) {
+ if (replName.equals(name)) {
+ return replName;
+ }
+ }
+
+ if (replName.contains("spark.")) {
+ replName = replName.replace("spark.", "");
+ }
+ return interpreterClassMap.get(replName);
+ }
+
+ /**
+ * General function to register hook event
+ * @param event The type of event to hook to (pre_exec, post_exec)
+ * @param cmd The code to be executed by the interpreter on given event
+ * @param replName Name of the interpreter
+ */
+ @Experimental
+ public void registerHook(String event, String cmd, String replName) {
+ String noteId = interpreterContext.getNoteId();
+ String className = getClassNameFromReplName(replName);
+ hooks.register(noteId, className, event, cmd);
+ }
+
+ /**
+ * registerHook() wrapper for current repl
+ * @param event The type of event to hook to (pre_exec, post_exec)
+ * @param cmd The code to be executed by the interpreter on given event
+ */
+ @Experimental
+ public void registerHook(String event, String cmd) {
+ String className = interpreterContext.getClassName();
+ registerHook(event, cmd, className);
+ }
+
+ /**
+ * Get the hook code
+ * @param event The type of event to hook to (pre_exec, post_exec)
+ * @param replName Name of the interpreter
+ */
+ @Experimental
+ public String getHook(String event, String replName) {
+ String noteId = interpreterContext.getNoteId();
+ String className = getClassNameFromReplName(replName);
+ return hooks.get(noteId, className, event);
+ }
+
+ /**
+ * getHook() wrapper for current repl
+ * @param event The type of event to hook to (pre_exec, post_exec)
+ */
+ @Experimental
+ public String getHook(String event) {
+ String className = interpreterContext.getClassName();
+ return getHook(event, className);
+ }
+
+ /**
+ * Unbind code from given hook event
+ * @param event The type of event to hook to (pre_exec, post_exec)
+ * @param replName Name of the interpreter
+ */
+ @Experimental
+ public void unregisterHook(String event, String replName) {
+ String noteId = interpreterContext.getNoteId();
+ String className = getClassNameFromReplName(replName);
+ hooks.unregister(noteId, className, event);
+ }
+
+ /**
+ * unregisterHook() wrapper for current repl
+ * @param event The type of event to hook to (pre_exec, post_exec)
+ */
+ @Experimental
+ public void unregisterHook(String event) {
+ String className = interpreterContext.getClassName();
+ unregisterHook(event, className);
+ }
/**
* Add object into resource pool
diff --git a/spark/src/main/java/org/apache/zeppelin/spark/ZeppelinR.java b/spark/src/main/java/org/apache/zeppelin/spark/ZeppelinR.java
index 26488337b11..961793db17e 100644
--- a/spark/src/main/java/org/apache/zeppelin/spark/ZeppelinR.java
+++ b/spark/src/main/java/org/apache/zeppelin/spark/ZeppelinR.java
@@ -141,6 +141,9 @@ public void open() throws IOException {
cmd.addArgument(Integer.toString(port));
cmd.addArgument(libPath);
cmd.addArgument(Integer.toString(sparkVersion.toNumber()));
+
+ // dump out the R command to facilitate manually running it, e.g. for fault diagnosis purposes
+ logger.debug(cmd.toString());
executor = new DefaultExecutor();
outputStream = new SparkOutputStream(logger);
diff --git a/spark/src/main/resources/python/zeppelin_pyspark.py b/spark/src/main/resources/python/zeppelin_pyspark.py
index 49e60d4577d..a53f503b83c 100644
--- a/spark/src/main/resources/python/zeppelin_pyspark.py
+++ b/spark/src/main/resources/python/zeppelin_pyspark.py
@@ -80,16 +80,16 @@ def put(self, key, value):
def get(self, key):
return self.__getitem__(key)
- def input(self, name, defaultValue = ""):
+ def input(self, name, defaultValue=""):
return self.z.input(name, defaultValue)
- def select(self, name, options, defaultValue = ""):
+ def select(self, name, options, defaultValue=""):
# auto_convert to ArrayList doesn't match the method signature on JVM side
tuples = list(map(lambda items: self.__tupleToScalaTuple2(items), options))
iterables = gateway.jvm.scala.collection.JavaConversions.collectionAsScalaIterable(tuples)
return self.z.select(name, defaultValue, iterables)
- def checkbox(self, name, options, defaultChecked = None):
+ def checkbox(self, name, options, defaultChecked=None):
if defaultChecked is None:
defaultChecked = list(map(lambda items: items[0], options))
optionTuples = list(map(lambda items: self.__tupleToScalaTuple2(items), options))
@@ -99,6 +99,23 @@ def checkbox(self, name, options, defaultChecked = None):
checkedIterables = self.z.checkbox(name, defaultCheckedIterables, optionIterables)
return gateway.jvm.scala.collection.JavaConversions.asJavaCollection(checkedIterables)
+ def registerHook(self, event, cmd, replName=None):
+ if replName is None:
+ self.z.registerHook(event, cmd)
+ else:
+ self.z.registerHook(event, cmd, replName)
+
+ def unregisterHook(self, event, replName=None):
+ if replName is None:
+ self.z.unregisterHook(event)
+ else:
+ self.z.unregisterHook(event, replName)
+
+ def getHook(self, event, replName=None):
+ if replName is None:
+ return self.z.getHook(event)
+ return self.z.getHook(event, replName)
+
def __tupleToScalaTuple2(self, tuple):
if (len(tuple) == 2):
return gateway.jvm.scala.Tuple2(tuple[0], tuple[1])
diff --git a/zeppelin-distribution/src/bin_license/LICENSE b/zeppelin-distribution/src/bin_license/LICENSE
index 2ee668a130e..99853d77c62 100644
--- a/zeppelin-distribution/src/bin_license/LICENSE
+++ b/zeppelin-distribution/src/bin_license/LICENSE
@@ -116,7 +116,6 @@ The following components are provided under Apache License.
(Apache 2.0) Utility classes for Jetty (org.mortbay.jetty:jetty-util:6.1.26 - http://javadox.com/org.mortbay.jetty/jetty/6.1.26/overview-tree.html)
(Apache 2.0) Servlet API (org.mortbay.jetty:servlet-api:2.5-20081211 - https://en.wikipedia.org/wiki/Jetty_(web_server))
(Apache 2.0) Google HTTP Client Library for Java (com.google.http-client:google-http-client-jackson2:1.21.0 - https://github.com/google/google-http-java-client/tree/dev/google-http-client-jackson2)
- (Apache 2.0) angular-esri-map (https://github.com/Esri/angular-esri-map)
(Apache 2.0) pegdown (org.pegdown:pegdown:1.6.0 - https://github.com/sirthias/pegdown)
(Apache 2.0) parboiled-java (org.parboiled:parboiled-java:1.1.7 - https://github.com/sirthias/parboiled)
(Apache 2.0) parboiled-core (org.parboiled:parboiled-core:1.1.7 - https://github.com/sirthias/parboiled)
@@ -156,7 +155,15 @@ The following components are provided under Apache License.
(Apache 2.0) Tachyon Project Core (org.tachyonproject:tachyon:0.6.4 - http://tachyonproject.org/tachyon/)
(Apache 2.0) Tachyon Project Client (org.tachyonproject:tachyon-client:0.6.4 - http://tachyonproject.org/tachyon-client/)
(Apache 2.0) javax.inject (javax.inject:javax.inject:1 - http://code.google.com/p/atinject/)
-
+ (Apache 2.0) Apache Pig (org.apache.pig:0.16 - http://pig.apache.org)
+ (Apache 2.0) tez-api (org.apache.tez:tez-api:0.7.0 - http://tez.apache.org)
+ (Apache 2.0) tez-common (org.apache.tez:tez-common:0.7.0 - http://tez.apache.org)
+ (Apache 2.0) tez-dag (org.apache.tez:tez-dag:0.7.0 - http://tez.apache.org)
+ (Apache 2.0) tez-runtime-library (org.apache.tez:runtime-library:0.7.0 - http://tez.apache.org)
+ (Apache 2.0) tez-runtime-internals (org.apache.tez:tez-runtime-internals:0.7.0 - http://tez.apache.org)
+ (Apache 2.0) tez-mapreduce (org.apache.tez:tez-mapreduce:0.7.0 - http://tez.apache.org)
+ (Apache 2.0) tez-yarn-timeline-history-with-acls (org.apache.tez:tez-yarn-timeline-history-with-acls:0.7.0 - http://tez.apache.org)
+
========================================================================
MIT licenses
========================================================================
@@ -252,7 +259,7 @@ The following components are provided under the BSD-style License.
(BSD-like) ASM asm-utils (org.ow2.asm:asm-utils:5.0.3 - http://asm.ow2.org/) - Copyright (c) 2000-2011 INRIA, France Telecom
(New BSD License) Markdown4j (org.commonjava.googlecode.markdown4j:markdown4j:jar:2.2-cj-1.0 - https://code.google.com/p/markdown4j/)
(New BSD License) Py4J (net.sf.py4j:py4j:0.9 - http://py4j.sourceforge.net/)
- (New BSD License) Py4J (net.sf.py4j:py4j:0.10.1 - http://py4j.sourceforge.net/) - https://github.com/bartdag/py4j/blob/0.10.1/LICENSE.txt
+ (New BSD License) Py4J (net.sf.py4j:py4j:0.10.3 - http://py4j.sourceforge.net/) - https://github.com/bartdag/py4j/blob/0.10.3/LICENSE.txt
(New BSD License) Markdown4j (org.commonjava.googlecode.markdown4j:markdown4j:jar:2.2-cj-1.0 - https://code.google.com/p/markdown4j/)
(BSD 3 Clause) Paranamer (com.thoughtworks.paranamer:paranamer:jar:2.6) - https://github.com/paul-hammant/paranamer/blob/paranamer-parent-2.6/LICENSE.txt
(BSD 3 Clause) netlib core (com.github.fommil.netlib:core:1.1.2 - https://github.com/fommil/netlib-java/core)
diff --git a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/Interpreter.java b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/Interpreter.java
index 9678b4691df..3e323200f6e 100644
--- a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/Interpreter.java
+++ b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/Interpreter.java
@@ -27,6 +27,7 @@
import com.google.gson.annotations.SerializedName;
import org.apache.zeppelin.annotation.ZeppelinApi;
+import org.apache.zeppelin.annotation.Experimental;
import org.apache.zeppelin.interpreter.thrift.InterpreterCompletion;
import org.apache.zeppelin.scheduler.Scheduler;
import org.apache.zeppelin.scheduler.SchedulerFactory;
@@ -203,6 +204,71 @@ public void setClassloaderUrls(URL[] classloaderUrls) {
this.classloaderUrls = classloaderUrls;
}
+ /**
+ * General function to register hook event
+ * @param noteId - Note to bind hook to
+ * @param event The type of event to hook to (pre_exec, post_exec)
+ * @param cmd The code to be executed by the interpreter on given event
+ */
+ @Experimental
+ public void registerHook(String noteId, String event, String cmd) {
+ InterpreterHookRegistry hooks = interpreterGroup.getInterpreterHookRegistry();
+ String className = getClassName();
+ hooks.register(noteId, className, event, cmd);
+ }
+
+ /**
+ * registerHook() wrapper for global scope
+ * @param event The type of event to hook to (pre_exec, post_exec)
+ * @param cmd The code to be executed by the interpreter on given event
+ */
+ @Experimental
+ public void registerHook(String event, String cmd) {
+ registerHook(null, event, cmd);
+ }
+
+ /**
+ * Get the hook code
+ * @param noteId - Note to bind hook to
+ * @param event The type of event to hook to (pre_exec, post_exec)
+ */
+ @Experimental
+ public String getHook(String noteId, String event) {
+ InterpreterHookRegistry hooks = interpreterGroup.getInterpreterHookRegistry();
+ String className = getClassName();
+ return hooks.get(noteId, className, event);
+ }
+
+ /**
+ * getHook() wrapper for global scope
+ * @param event The type of event to hook to (pre_exec, post_exec)
+ */
+ @Experimental
+ public String getHook(String event) {
+ return getHook(null, event);
+ }
+
+ /**
+ * Unbind code from given hook event
+ * @param noteId - Note to bind hook to
+ * @param event The type of event to hook to (pre_exec, post_exec)
+ */
+ @Experimental
+ public void unregisterHook(String noteId, String event) {
+ InterpreterHookRegistry hooks = interpreterGroup.getInterpreterHookRegistry();
+ String className = getClassName();
+ hooks.unregister(noteId, className, event);
+ }
+
+ /**
+ * unregisterHook() wrapper for global scope
+ * @param event The type of event to hook to (pre_exec, post_exec)
+ */
+ @Experimental
+ public void unregisterHook(String event) {
+ unregisterHook(null, event);
+ }
+
@ZeppelinApi
public Interpreter getInterpreterInTheSameSessionByClassName(String className) {
synchronized (interpreterGroup) {
diff --git a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/InterpreterContext.java b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/InterpreterContext.java
index 21ca2e67b72..e33b9352252 100644
--- a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/InterpreterContext.java
+++ b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/InterpreterContext.java
@@ -57,6 +57,7 @@ public static void remove() {
private AngularObjectRegistry angularObjectRegistry;
private ResourcePool resourcePool;
private List runners;
+ private String className;
public InterpreterContext(String noteId,
String paragraphId,
@@ -124,4 +125,11 @@ public List getRunners() {
return runners;
}
+ public String getClassName() {
+ return className;
+ }
+
+ public void setClassName(String className) {
+ this.className = className;
+ }
}
diff --git a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/InterpreterGroup.java b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/InterpreterGroup.java
index bc56784b15b..f3b158c26a6 100644
--- a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/InterpreterGroup.java
+++ b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/InterpreterGroup.java
@@ -33,7 +33,7 @@
* and InterpreterGroup will have reference to these all interpreters.
*
* Remember, list of interpreters are dedicated to a note.
- * (when InterpreterOption.perNoteSession==true)
+ * (when InterpreterOption.session==true)
* So InterpreterGroup internally manages map of [noteId, list of interpreters]
*
* A InterpreterGroup runs on interpreter process.
@@ -45,6 +45,7 @@ public class InterpreterGroup extends ConcurrentHashMap intpForNote = this.get(noteId);
destroy(intpForNote);
+
+ if (remoteInterpreterProcess != null) {
+ remoteInterpreterProcess.dereference();
+ if (remoteInterpreterProcess.referenceCount() <= 0) {
+ remoteInterpreterProcess = null;
+ allInterpreterGroups.remove(id);
+ }
+ }
}
@@ -213,6 +230,7 @@ public void destroy() {
while (remoteInterpreterProcess.referenceCount() > 0) {
remoteInterpreterProcess.dereference();
}
+ remoteInterpreterProcess = null;
}
allInterpreterGroups.remove(id);
diff --git a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/InterpreterHookListener.java b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/InterpreterHookListener.java
new file mode 100644
index 00000000000..c70212c7b7e
--- /dev/null
+++ b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/InterpreterHookListener.java
@@ -0,0 +1,33 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.zeppelin.interpreter;
+
+/**
+ * An interface for processing custom callback code into the interpreter.
+ */
+public interface InterpreterHookListener {
+ /**
+ * Prepends pre-execute hook code to the script that will be interpreted
+ */
+ public void onPreExecute(String script);
+
+ /**
+ * Appends post-execute hook code to the script that will be interpreted
+ */
+ public void onPostExecute(String script);
+}
diff --git a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/InterpreterHookRegistry.java b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/InterpreterHookRegistry.java
new file mode 100644
index 00000000000..0917775d692
--- /dev/null
+++ b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/InterpreterHookRegistry.java
@@ -0,0 +1,156 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.zeppelin.interpreter;
+
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * The InterpreterinterpreterHookRegistry specifies code to be conditionally executed by an
+ * interpreter. The constants defined in this class denote currently
+ * supported events. Each instance is bound to a single InterpreterGroup.
+ * Scope is determined on a per-note basis (except when null for global scope).
+ */
+public class InterpreterHookRegistry {
+ public static final String GLOBAL_KEY = "_GLOBAL_";
+ private String interpreterId;
+ private Map>> registry =
+ new HashMap>>();
+
+ /**
+ * hookRegistry constructor.
+ *
+ * @param interpreterId The Id of the InterpreterGroup instance to bind to
+ */
+ public InterpreterHookRegistry(final String interpreterId) {
+ this.interpreterId = interpreterId;
+ }
+
+ /**
+ * Get the interpreterGroup id this instance is bound to
+ */
+ public String getInterpreterId() {
+ return interpreterId;
+ }
+
+ /**
+ * Adds a note to the registry
+ *
+ * @param noteId The Id of the Note instance to add
+ */
+ public void addNote(String noteId) {
+ synchronized (registry) {
+ if (registry.get(noteId) == null) {
+ registry.put(noteId, new HashMap>());
+ }
+ }
+ }
+
+ /**
+ * Adds a className to the registry
+ *
+ * @param noteId The note id
+ * @param className The name of the interpreter repl to map the hooks to
+ */
+ public void addRepl(String noteId, String className) {
+ synchronized (registry) {
+ addNote(noteId);
+ if (registry.get(noteId).get(className) == null) {
+ registry.get(noteId).put(className, new HashMap());
+ }
+ }
+ }
+
+ /**
+ * Register a hook for a specific event.
+ *
+ * @param noteId Denotes the note this instance belongs to
+ * @param className The name of the interpreter repl to map the hooks to
+ * @param event hook event (see constants defined in this class)
+ * @param cmd Code to be executed by the interpreter
+ */
+ public void register(String noteId, String className,
+ String event, String cmd) throws IllegalArgumentException {
+ synchronized (registry) {
+ if (noteId == null) {
+ noteId = GLOBAL_KEY;
+ }
+ addRepl(noteId, className);
+ if (!event.equals(HookType.POST_EXEC) && !event.equals(HookType.PRE_EXEC) &&
+ !event.equals(HookType.POST_EXEC_DEV) && !event.equals(HookType.PRE_EXEC_DEV)) {
+ throw new IllegalArgumentException("Must be " + HookType.POST_EXEC + ", " +
+ HookType.POST_EXEC_DEV + ", " +
+ HookType.PRE_EXEC + " or " +
+ HookType.PRE_EXEC_DEV);
+ }
+ registry.get(noteId).get(className).put(event, cmd);
+ }
+ }
+
+ /**
+ * Unregister a hook for a specific event.
+ *
+ * @param noteId Denotes the note this instance belongs to
+ * @param className The name of the interpreter repl to map the hooks to
+ * @param event hook event (see constants defined in this class)
+ */
+ public void unregister(String noteId, String className, String event) {
+ synchronized (registry) {
+ if (noteId == null) {
+ noteId = GLOBAL_KEY;
+ }
+ addRepl(noteId, className);
+ registry.get(noteId).get(className).remove(event);
+ }
+ }
+
+ /**
+ * Get a hook for a specific event.
+ *
+ * @param noteId Denotes the note this instance belongs to
+ * @param className The name of the interpreter repl to map the hooks to
+ * @param event hook event (see constants defined in this class)
+ */
+ public String get(String noteId, String className, String event) {
+ synchronized (registry) {
+ if (noteId == null) {
+ noteId = GLOBAL_KEY;
+ }
+ addRepl(noteId, className);
+ return registry.get(noteId).get(className).get(event);
+ }
+ }
+
+ /**
+ * Container for hook event type constants
+ */
+ public static final class HookType {
+ // Execute the hook code PRIOR to main paragraph code execution
+ public static final String PRE_EXEC = "pre_exec";
+
+ // Execute the hook code AFTER main paragraph code execution
+ public static final String POST_EXEC = "post_exec";
+
+ // Same as above but reserved for interpreter developers, in order to allow
+ // notebook users to use the above without overwriting registry settings
+ // that are initialized directly in subclasses of Interpreter.
+ public static final String PRE_EXEC_DEV = "pre_exec_dev";
+ public static final String POST_EXEC_DEV = "post_exec_dev";
+ }
+
+}
diff --git a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/LazyOpenInterpreter.java b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/LazyOpenInterpreter.java
index c62ab05eb1f..425ae20a4f1 100644
--- a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/LazyOpenInterpreter.java
+++ b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/LazyOpenInterpreter.java
@@ -147,4 +147,34 @@ public void setInterpreterGroup(InterpreterGroup interpreterGroup) {
public void setClassloaderUrls(URL [] urls) {
intp.setClassloaderUrls(urls);
}
+
+ @Override
+ public void registerHook(String noteId, String event, String cmd) {
+ intp.registerHook(noteId, event, cmd);
+ }
+
+ @Override
+ public void registerHook(String event, String cmd) {
+ intp.registerHook(event, cmd);
+ }
+
+ @Override
+ public String getHook(String noteId, String event) {
+ return intp.getHook(noteId, event);
+ }
+
+ @Override
+ public String getHook(String event) {
+ return intp.getHook(event);
+ }
+
+ @Override
+ public void unregisterHook(String noteId, String event) {
+ intp.unregisterHook(noteId, event);
+ }
+
+ @Override
+ public void unregisterHook(String event) {
+ intp.unregisterHook(event);
+ }
}
diff --git a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreter.java b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreter.java
index 073b84bbda2..e0cdaa338b1 100644
--- a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreter.java
+++ b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreter.java
@@ -298,6 +298,7 @@ public InterpreterResult interpret(String st, InterpreterContext context) {
if (logger.isDebugEnabled()) {
logger.debug("st:\n{}", st);
}
+
FormType form = getFormType();
RemoteInterpreterProcess interpreterProcess = getInterpreterProcess();
Client client = null;
diff --git a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreterServer.java b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreterServer.java
index 8344366e569..0a7b1ed6912 100644
--- a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreterServer.java
+++ b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreterServer.java
@@ -33,6 +33,8 @@
import org.apache.zeppelin.display.*;
import org.apache.zeppelin.helium.*;
import org.apache.zeppelin.interpreter.*;
+import org.apache.zeppelin.interpreter.InterpreterHookRegistry.HookType;
+import org.apache.zeppelin.interpreter.InterpreterHookListener;
import org.apache.zeppelin.interpreter.InterpreterResult.Code;
import org.apache.zeppelin.interpreter.dev.ZeppelinDevServer;
import org.apache.zeppelin.interpreter.thrift.*;
@@ -60,6 +62,7 @@ public class RemoteInterpreterServer
InterpreterGroup interpreterGroup;
AngularObjectRegistry angularObjectRegistry;
+ InterpreterHookRegistry hookRegistry;
DistributedResourcePool resourcePool;
private ApplicationLoader appLoader;
@@ -152,7 +155,9 @@ public void createInterpreter(String interpreterGroupId, String noteId, String
if (interpreterGroup == null) {
interpreterGroup = new InterpreterGroup(interpreterGroupId);
angularObjectRegistry = new AngularObjectRegistry(interpreterGroup.getId(), this);
+ hookRegistry = new InterpreterHookRegistry(interpreterGroup.getId());
resourcePool = new DistributedResourcePool(interpreterGroup.getId(), eventClient);
+ interpreterGroup.setInterpreterHookRegistry(hookRegistry);
interpreterGroup.setAngularObjectRegistry(angularObjectRegistry);
interpreterGroup.setResourcePool(resourcePool);
@@ -290,6 +295,7 @@ public RemoteInterpreterResult interpret(String noteId, String className, String
}
Interpreter intp = getInterpreter(noteId, className);
InterpreterContext context = convert(interpreterContext);
+ context.setClassName(intp.getClassName());
Scheduler scheduler = intp.getScheduler();
InterpretJobListener jobListener = new InterpretJobListener();
@@ -383,10 +389,61 @@ public Map info() {
return infos;
}
+ private void processInterpreterHooks(final String noteId) {
+ InterpreterHookListener hookListener = new InterpreterHookListener() {
+ @Override
+ public void onPreExecute(String script) {
+ String cmdDev = interpreter.getHook(noteId, HookType.PRE_EXEC_DEV);
+ String cmdUser = interpreter.getHook(noteId, HookType.PRE_EXEC);
+
+ // User defined hook should be executed before dev hook
+ List cmds = Arrays.asList(cmdDev, cmdUser);
+ for (String cmd : cmds) {
+ if (cmd != null) {
+ script = cmd + '\n' + script;
+ }
+ }
+
+ InterpretJob.this.script = script;
+ }
+
+ @Override
+ public void onPostExecute(String script) {
+ String cmdDev = interpreter.getHook(noteId, HookType.POST_EXEC_DEV);
+ String cmdUser = interpreter.getHook(noteId, HookType.POST_EXEC);
+
+ // User defined hook should be executed after dev hook
+ List cmds = Arrays.asList(cmdUser, cmdDev);
+ for (String cmd : cmds) {
+ if (cmd != null) {
+ script += '\n' + cmd;
+ }
+ }
+
+ InterpretJob.this.script = script;
+ }
+ };
+ hookListener.onPreExecute(script);
+ hookListener.onPostExecute(script);
+ }
+
@Override
protected Object jobRun() throws Throwable {
try {
InterpreterContext.set(context);
+
+ // Open the interpreter instance prior to calling interpret().
+ // This is necessary because the earliest we can register a hook
+ // is from within the open() method.
+ LazyOpenInterpreter lazy = (LazyOpenInterpreter) interpreter;
+ if (!lazy.isOpen()) {
+ lazy.open();
+ }
+
+ // Add hooks to script from registry.
+ // Global scope first, followed by notebook scope
+ processInterpreterHooks(null);
+ processInterpreterHooks(context.getNoteId());
InterpreterResult result = interpreter.interpret(script, context);
// data from context.out is prepended to InterpreterResult if both defined
diff --git a/zeppelin-interpreter/src/test/java/org/apache/zeppelin/interpreter/InterpreterHookRegistryTest.java b/zeppelin-interpreter/src/test/java/org/apache/zeppelin/interpreter/InterpreterHookRegistryTest.java
new file mode 100644
index 00000000000..7614e9eb204
--- /dev/null
+++ b/zeppelin-interpreter/src/test/java/org/apache/zeppelin/interpreter/InterpreterHookRegistryTest.java
@@ -0,0 +1,77 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.zeppelin.interpreter;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNull;
+
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.junit.Test;
+
+public class InterpreterHookRegistryTest {
+
+ @Test
+ public void testBasic() {
+ final String PRE_EXEC = InterpreterHookRegistry.HookType.PRE_EXEC;
+ final String POST_EXEC = InterpreterHookRegistry.HookType.POST_EXEC;
+ final String PRE_EXEC_DEV = InterpreterHookRegistry.HookType.PRE_EXEC_DEV;
+ final String POST_EXEC_DEV = InterpreterHookRegistry.HookType.POST_EXEC_DEV;
+ final String GLOBAL_KEY = InterpreterHookRegistry.GLOBAL_KEY;
+ final String noteId = "note";
+ final String className = "class";
+ final String preExecHook = "pre";
+ final String postExecHook = "post";
+ InterpreterHookRegistry registry = new InterpreterHookRegistry("intpId");
+
+ // Test register()
+ registry.register(noteId, className, PRE_EXEC, preExecHook);
+ registry.register(noteId, className, POST_EXEC, postExecHook);
+ registry.register(noteId, className, PRE_EXEC_DEV, preExecHook);
+ registry.register(noteId, className, POST_EXEC_DEV, postExecHook);
+
+ // Test get()
+ assertEquals(registry.get(noteId, className, PRE_EXEC), preExecHook);
+ assertEquals(registry.get(noteId, className, POST_EXEC), postExecHook);
+ assertEquals(registry.get(noteId, className, PRE_EXEC_DEV), preExecHook);
+ assertEquals(registry.get(noteId, className, POST_EXEC_DEV), postExecHook);
+
+ // Test Unregister
+ registry.unregister(noteId, className, PRE_EXEC);
+ registry.unregister(noteId, className, POST_EXEC);
+ registry.unregister(noteId, className, PRE_EXEC_DEV);
+ registry.unregister(noteId, className, POST_EXEC_DEV);
+ assertNull(registry.get(noteId, className, PRE_EXEC));
+ assertNull(registry.get(noteId, className, POST_EXEC));
+ assertNull(registry.get(noteId, className, PRE_EXEC_DEV));
+ assertNull(registry.get(noteId, className, POST_EXEC_DEV));
+
+ // Test Global Scope
+ registry.register(null, className, PRE_EXEC, preExecHook);
+ assertEquals(registry.get(GLOBAL_KEY, className, PRE_EXEC), preExecHook);
+ }
+
+ @Test(expected = IllegalArgumentException.class)
+ public void testValidEventCode() {
+ InterpreterHookRegistry registry = new InterpreterHookRegistry("intpId");
+
+ // Test that only valid event codes ("pre_exec", "post_exec") are accepted
+ registry.register("foo", "bar", "baz", "whatever");
+ }
+
+}
diff --git a/zeppelin-server/src/main/java/org/apache/zeppelin/rest/NotebookRestApi.java b/zeppelin-server/src/main/java/org/apache/zeppelin/rest/NotebookRestApi.java
index 727211292b2..d5814046183 100644
--- a/zeppelin-server/src/main/java/org/apache/zeppelin/rest/NotebookRestApi.java
+++ b/zeppelin-server/src/main/java/org/apache/zeppelin/rest/NotebookRestApi.java
@@ -39,7 +39,6 @@
import com.google.gson.Gson;
import org.apache.commons.lang3.StringUtils;
import org.apache.zeppelin.interpreter.InterpreterResult;
-import org.apache.zeppelin.scheduler.Job;
import org.apache.zeppelin.utils.InterpreterBindingUtils;
import org.quartz.CronExpression;
import org.slf4j.Logger;
@@ -162,6 +161,7 @@ public Response putNotePermissions(@PathParam("noteId") String noteId, String re
AuthenticationInfo subject = new AuthenticationInfo(SecurityUtils.getPrincipal());
note.persist(subject);
notebookServer.broadcastNote(note);
+ notebookServer.broadcastNoteList(subject, userAndRoles);
return new JsonResponse<>(Status.OK).build();
}
@@ -176,7 +176,7 @@ public Response putNotePermissions(@PathParam("noteId") String noteId, String re
public Response bind(@PathParam("noteId") String noteId, String req) throws IOException {
List settingIdList = gson.fromJson(req, new TypeToken>() {
}.getType());
- notebook.bindInterpretersToNote(noteId, settingIdList);
+ notebook.bindInterpretersToNote(SecurityUtils.getPrincipal(), noteId, settingIdList);
return new JsonResponse<>(Status.OK).build();
}
@@ -198,7 +198,8 @@ public Response bind(@PathParam("noteId") String noteId) {
@ZeppelinApi
public Response getNotebookList() throws IOException {
AuthenticationInfo subject = new AuthenticationInfo(SecurityUtils.getPrincipal());
- List> notesInfo = notebookServer.generateNotebooksInfo(false, subject);
+ List> notesInfo = notebookServer.generateNotebooksInfo(false, subject,
+ SecurityUtils.getRoles());
return new JsonResponse<>(Status.OK, "", notesInfo).build();
}
@@ -277,7 +278,7 @@ public Response createNote(String message) throws IOException {
note.setName(noteName);
note.persist(subject);
notebookServer.broadcastNote(note);
- notebookServer.broadcastNoteList(subject);
+ notebookServer.broadcastNoteList(subject, SecurityUtils.getRoles());
return new JsonResponse<>(Status.CREATED, "", note.getId()).build();
}
@@ -301,7 +302,7 @@ public Response deleteNote(@PathParam("notebookId") String notebookId) throws IO
}
}
- notebookServer.broadcastNoteList(subject);
+ notebookServer.broadcastNoteList(subject, SecurityUtils.getRoles());
return new JsonResponse<>(Status.OK, "").build();
}
@@ -326,7 +327,7 @@ public Response cloneNote(@PathParam("notebookId") String notebookId, String mes
AuthenticationInfo subject = new AuthenticationInfo(SecurityUtils.getPrincipal());
Note newNote = notebook.cloneNote(notebookId, newNoteName, subject);
notebookServer.broadcastNote(newNote);
- notebookServer.broadcastNoteList(subject);
+ notebookServer.broadcastNoteList(subject, SecurityUtils.getRoles());
return new JsonResponse<>(Status.CREATED, "", newNote.getId()).build();
}
@@ -457,7 +458,7 @@ public Response deleteParagraph(@PathParam("notebookId") String notebookId,
}
AuthenticationInfo subject = new AuthenticationInfo(SecurityUtils.getPrincipal());
- note.removeParagraph(paragraphId);
+ note.removeParagraph(SecurityUtils.getPrincipal(), paragraphId);
note.persist(subject);
notebookServer.broadcastNote(note);
@@ -598,6 +599,11 @@ public Response runParagraph(@PathParam("notebookId") String notebookId,
// handle params if presented
handleParagraphParams(message, note, paragraph);
+ AuthenticationInfo subject = new AuthenticationInfo(SecurityUtils.getPrincipal());
+
+ paragraph.setAuthenticationInfo(subject);
+ note.persist(subject);
+
note.run(paragraph.getId());
return new JsonResponse<>(Status.OK).build();
}
diff --git a/zeppelin-server/src/main/java/org/apache/zeppelin/server/ZeppelinServer.java b/zeppelin-server/src/main/java/org/apache/zeppelin/server/ZeppelinServer.java
index 415abd73708..de02fe0f3c8 100644
--- a/zeppelin-server/src/main/java/org/apache/zeppelin/server/ZeppelinServer.java
+++ b/zeppelin-server/src/main/java/org/apache/zeppelin/server/ZeppelinServer.java
@@ -89,10 +89,10 @@ public ZeppelinServer() throws Exception {
this.heliumApplicationFactory = new HeliumApplicationFactory();
this.schedulerFactory = new SchedulerFactory();
this.replFactory = new InterpreterFactory(conf, notebookWsServer,
- notebookWsServer, heliumApplicationFactory, depResolver);
+ notebookWsServer, heliumApplicationFactory, depResolver, SecurityUtils.isAuthenticated());
this.notebookRepo = new NotebookRepoSync(conf);
this.notebookIndex = new LuceneSearch();
- this.notebookAuthorization = new NotebookAuthorization(conf);
+ this.notebookAuthorization = NotebookAuthorization.init(conf);
this.credentials = new Credentials(conf.credentialsPersist(), conf.getCredentialsPath());
notebook = new Notebook(conf,
notebookRepo, schedulerFactory, replFactory, notebookWsServer,
diff --git a/zeppelin-server/src/main/java/org/apache/zeppelin/socket/NotebookServer.java b/zeppelin-server/src/main/java/org/apache/zeppelin/socket/NotebookServer.java
index ab7394b6208..2f9faba42db 100644
--- a/zeppelin-server/src/main/java/org/apache/zeppelin/socket/NotebookServer.java
+++ b/zeppelin-server/src/main/java/org/apache/zeppelin/socket/NotebookServer.java
@@ -20,7 +20,6 @@
import com.google.gson.Gson;
import com.google.gson.GsonBuilder;
import com.google.gson.reflect.TypeToken;
-
import org.apache.commons.lang.StringUtils;
import org.apache.commons.vfs2.FileSystemException;
import org.apache.zeppelin.conf.ZeppelinConfiguration;
@@ -60,6 +59,7 @@
import java.net.URISyntaxException;
import java.net.UnknownHostException;
import java.util.*;
+import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
/**
@@ -86,6 +86,8 @@ String getKey() {
Gson gson = new GsonBuilder().setDateFormat("yyyy-MM-dd'T'HH:mm:ssZ").create();
final Map> noteSocketMap = new HashMap<>();
final Queue connectedSockets = new ConcurrentLinkedQueue<>();
+ final Map> userConnectedSockets =
+ new ConcurrentHashMap>();
private Notebook notebook() {
return ZeppelinServer.notebook;
@@ -131,7 +133,7 @@ public void onMessage(NotebookSocket conn, String msg) {
if (LOG.isTraceEnabled()) {
LOG.trace("RECEIVE MSG = " + messagereceived);
}
-
+
String ticket = TicketContainer.instance.getTicket(messagereceived.principal);
if (ticket != null && !ticket.equals(messagereceived.ticket)){
/* not to pollute logs, log instead of exception */
@@ -161,15 +163,18 @@ public void onMessage(NotebookSocket conn, String msg) {
userAndRoles.addAll(roles);
}
}
+ if (StringUtils.isEmpty(conn.getUser())) {
+ addUserConnection(messagereceived.principal, conn);
+ }
AuthenticationInfo subject = new AuthenticationInfo(messagereceived.principal);
/** Lets be elegant here */
switch (messagereceived.op) {
case LIST_NOTES:
- unicastNoteList(conn, subject);
+ unicastNoteList(conn, subject, userAndRoles);
break;
case RELOAD_NOTES_FROM_REPO:
- broadcastReloadedNoteList(subject);
+ broadcastReloadedNoteList(subject, userAndRoles);
break;
case GET_HOME_NOTE:
sendHomeNote(conn, userAndRoles, notebook, messagereceived);
@@ -268,6 +273,26 @@ public void onClose(NotebookSocket conn, int code, String reason) {
.getRemoteAddr(), conn.getRequest().getRemotePort(), code, reason);
removeConnectionFromAllNote(conn);
connectedSockets.remove(conn);
+ removeUserConnection(conn.getUser(), conn);
+ }
+
+ private void removeUserConnection(String user, NotebookSocket conn) {
+ if (userConnectedSockets.containsKey(user)) {
+ userConnectedSockets.get(user).remove(conn);
+ } else {
+ LOG.warn("Closing connection that is absent in user connections");
+ }
+ }
+
+ private void addUserConnection(String user, NotebookSocket conn) {
+ conn.setUser(user);
+ if (userConnectedSockets.containsKey(user)) {
+ userConnectedSockets.get(user).add(conn);
+ } else {
+ Queue socketQueue = new ConcurrentLinkedQueue<>();
+ socketQueue.add(conn);
+ userConnectedSockets.put(user, socketQueue);
+ }
}
protected Message deserializeMessage(String msg) {
@@ -383,8 +408,12 @@ private void broadcastExcept(String noteId, Message m, NotebookSocket exclude) {
}
}
- private void broadcastAll(Message m) {
- for (NotebookSocket conn : connectedSockets) {
+ private void multicastToUser(String user, Message m) {
+ if (!userConnectedSockets.containsKey(user)) {
+ LOG.warn("Broadcasting to user that is not in connections map");
+ return;
+ }
+ for (NotebookSocket conn: userConnectedSockets.get(user)) {
try {
conn.send(serializeMessage(m));
} catch (IOException e) {
@@ -442,7 +471,8 @@ public void saveInterpreterBindings(NotebookSocket conn, Message fromMessage) {
List settingIdList = gson.fromJson(String.valueOf(
fromMessage.data.get("selectedSettingIds")), new TypeToken>() {
}.getType());
- notebook().bindInterpretersToNote(noteId, settingIdList);
+ AuthenticationInfo subject = new AuthenticationInfo(fromMessage.principal);
+ notebook().bindInterpretersToNote(subject.getUser(), noteId, settingIdList);
broadcastInterpreterBindings(noteId,
InterpreterBindingUtils.getInterpreterBindings(notebook(), noteId));
} catch (Exception e) {
@@ -460,7 +490,7 @@ public void getInterpreterBindings(NotebookSocket conn, Message fromMessage)
}
public List> generateNotebooksInfo(boolean needsReload,
- AuthenticationInfo subject) {
+ AuthenticationInfo subject, HashSet userAndRoles) {
Notebook notebook = notebook();
@@ -476,7 +506,8 @@ public List> generateNotebooksInfo(boolean needsReload,
LOG.error("Fail to reload notes from repository", e);
}
}
- List notes = notebook.getAllNotes(subject);
+
+ List notes = notebook.getAllNotes(userAndRoles);
List> notesInfo = new LinkedList<>();
for (Note note : notes) {
Map info = new HashMap<>();
@@ -503,19 +534,45 @@ public void broadcastInterpreterBindings(String noteId,
.put("interpreterBindings", settingList));
}
- public void broadcastNoteList(AuthenticationInfo subject) {
- List> notesInfo = generateNotebooksInfo(false, subject);
- broadcastAll(new Message(OP.NOTES_INFO).put("notes", notesInfo));
+ public void broadcastNoteList(AuthenticationInfo subject, HashSet userAndRoles) {
+ if (subject == null) {
+ subject = new AuthenticationInfo(StringUtils.EMPTY);
+ }
+ //send first to requesting user
+ List> notesInfo = generateNotebooksInfo(false, subject, userAndRoles);
+ multicastToUser(subject.getUser(), new Message(OP.NOTES_INFO).put("notes", notesInfo));
+ //to others afterwards
+ for (String user: userConnectedSockets.keySet()) {
+ if (subject.getUser() == user) {
+ continue;
+ }
+ notesInfo = generateNotebooksInfo(false, new AuthenticationInfo(user), userAndRoles);
+ multicastToUser(user, new Message(OP.NOTES_INFO).put("notes", notesInfo));
+ }
}
- public void unicastNoteList(NotebookSocket conn, AuthenticationInfo subject) {
- List> notesInfo = generateNotebooksInfo(false, subject);
+ public void unicastNoteList(NotebookSocket conn, AuthenticationInfo subject,
+ HashSet userAndRoles) {
+ List> notesInfo = generateNotebooksInfo(false, subject, userAndRoles);
unicast(new Message(OP.NOTES_INFO).put("notes", notesInfo), conn);
}
- public void broadcastReloadedNoteList(AuthenticationInfo subject) {
- List> notesInfo = generateNotebooksInfo(true, subject);
- broadcastAll(new Message(OP.NOTES_INFO).put("notes", notesInfo));
+ public void broadcastReloadedNoteList(AuthenticationInfo subject, HashSet userAndRoles) {
+ if (subject == null) {
+ subject = new AuthenticationInfo(StringUtils.EMPTY);
+ }
+ //reload and reply first to requesting user
+ List> notesInfo = generateNotebooksInfo(true, subject, userAndRoles);
+ multicastToUser(subject.getUser(), new Message(OP.NOTES_INFO).put("notes", notesInfo));
+ //to others afterwards
+ for (String user: userConnectedSockets.keySet()) {
+ if (subject.getUser() == user) {
+ continue;
+ }
+ //reloaded already above; parameter - false
+ notesInfo = generateNotebooksInfo(false, new AuthenticationInfo(user), userAndRoles);
+ multicastToUser(user, new Message(OP.NOTES_INFO).put("notes", notesInfo));
+ }
}
void permissionError(NotebookSocket conn, String op,
@@ -544,6 +601,8 @@ private void sendNote(NotebookSocket conn, HashSet userAndRoles, Noteboo
return;
}
+ String user = fromMessage.principal;
+
Note note = notebook.getNote(noteId);
NotebookAuthorization notebookAuthorization = notebook.getNotebookAuthorization();
if (note != null) {
@@ -554,7 +613,7 @@ private void sendNote(NotebookSocket conn, HashSet userAndRoles, Noteboo
}
addConnectionToNote(note.getId(), conn);
conn.send(serializeMessage(new Message(OP.NOTE).put("note", note)));
- sendAllAngularObjects(note, conn);
+ sendAllAngularObjects(note, user, conn);
} else {
conn.send(serializeMessage(new Message(OP.NOTE).put("note", null)));
}
@@ -563,6 +622,7 @@ private void sendNote(NotebookSocket conn, HashSet userAndRoles, Noteboo
private void sendHomeNote(NotebookSocket conn, HashSet userAndRoles,
Notebook notebook, Message fromMessage) throws IOException {
String noteId = notebook.getConf().getString(ConfVars.ZEPPELIN_NOTEBOOK_HOMESCREEN);
+ String user = fromMessage.principal;
Note note = null;
if (noteId != null) {
@@ -578,7 +638,7 @@ private void sendHomeNote(NotebookSocket conn, HashSet userAndRoles,
}
addConnectionToNote(note.getId(), conn);
conn.send(serializeMessage(new Message(OP.NOTE).put("note", note)));
- sendAllAngularObjects(note, conn);
+ sendAllAngularObjects(note, user, conn);
} else {
removeConnectionFromAllNote(conn);
conn.send(serializeMessage(new Message(OP.NOTE).put("note", null)));
@@ -618,7 +678,7 @@ private void updateNote(NotebookSocket conn, HashSet userAndRoles,
AuthenticationInfo subject = new AuthenticationInfo(fromMessage.principal);
note.persist(subject);
broadcastNote(note);
- broadcastNoteList(subject);
+ broadcastNoteList(subject, userAndRoles);
}
}
@@ -653,7 +713,7 @@ private void createNote(NotebookSocket conn, HashSet userAndRoles,
note.persist(subject);
addConnectionToNote(note.getId(), (NotebookSocket) conn);
conn.send(serializeMessage(new Message(OP.NEW_NOTE).put("note", note)));
- broadcastNoteList(subject);
+ broadcastNoteList(subject, userAndRoles);
}
private void removeNote(NotebookSocket conn, HashSet userAndRoles,
@@ -675,7 +735,7 @@ private void removeNote(NotebookSocket conn, HashSet userAndRoles,
AuthenticationInfo subject = new AuthenticationInfo(fromMessage.principal);
notebook.removeNote(noteId, subject);
removeNote(noteId);
- broadcastNoteList(subject);
+ broadcastNoteList(subject, userAndRoles);
}
private void updateParagraph(NotebookSocket conn, HashSet userAndRoles,
@@ -717,7 +777,7 @@ private void cloneNote(NotebookSocket conn, HashSet userAndRoles,
AuthenticationInfo subject = new AuthenticationInfo(fromMessage.principal);
addConnectionToNote(newNote.getId(), (NotebookSocket) conn);
conn.send(serializeMessage(new Message(OP.NEW_NOTE).put("note", newNote)));
- broadcastNoteList(subject);
+ broadcastNoteList(subject, userAndRoles);
}
protected Note importNote(NotebookSocket conn, HashSet userAndRoles,
@@ -730,11 +790,13 @@ protected Note importNote(NotebookSocket conn, HashSet userAndRoles,
AuthenticationInfo subject = null;
if (fromMessage.principal != null) {
subject = new AuthenticationInfo(fromMessage.principal);
+ } else {
+ subject = new AuthenticationInfo("anonymous");
}
note = notebook.importNote(noteJson, noteName, subject);
note.persist(subject);
broadcastNote(note);
- broadcastNoteList(subject);
+ broadcastNoteList(subject, userAndRoles);
}
return note;
}
@@ -748,7 +810,7 @@ private void removeParagraph(NotebookSocket conn, HashSet userAndRoles,
String noteId = getOpenNoteId(conn);
final Note note = notebook.getNote(noteId);
NotebookAuthorization notebookAuthorization = notebook.getNotebookAuthorization();
- AuthenticationInfo subject = new AuthenticationInfo(SecurityUtils.getPrincipal());
+ AuthenticationInfo subject = new AuthenticationInfo(fromMessage.principal);
if (!notebookAuthorization.isWriter(noteId, userAndRoles)) {
permissionError(conn, "write", fromMessage.principal,
userAndRoles, notebookAuthorization.getWriters(noteId));
@@ -757,7 +819,7 @@ private void removeParagraph(NotebookSocket conn, HashSet userAndRoles,
/** We dont want to remove the last paragraph */
if (!note.isLastParagraph(paragraphId)) {
- note.removeParagraph(paragraphId);
+ note.removeParagraph(subject.getUser(), paragraphId);
note.persist(subject);
broadcastNote(note);
}
@@ -813,6 +875,7 @@ private void angularObjectUpdated(NotebookSocket conn, HashSet userAndRo
String interpreterGroupId = (String) fromMessage.get("interpreterGroupId");
String varName = (String) fromMessage.get("name");
Object varValue = fromMessage.get("value");
+ String user = fromMessage.principal;
AngularObject ao = null;
boolean global = false;
// propagate change to (Remote) AngularObjectRegistry
@@ -821,12 +884,12 @@ private void angularObjectUpdated(NotebookSocket conn, HashSet userAndRo
List settings = notebook.getInterpreterFactory()
.getInterpreterSettings(note.getId());
for (InterpreterSetting setting : settings) {
- if (setting.getInterpreterGroup(note.getId()) == null) {
+ if (setting.getInterpreterGroup(user, note.getId()) == null) {
continue;
}
- if (interpreterGroupId.equals(setting.getInterpreterGroup(note.getId()).getId())) {
+ if (interpreterGroupId.equals(setting.getInterpreterGroup(user, note.getId()).getId())) {
AngularObjectRegistry angularObjectRegistry = setting
- .getInterpreterGroup(note.getId()).getAngularObjectRegistry();
+ .getInterpreterGroup(user, note.getId()).getAngularObjectRegistry();
// first trying to get local registry
ao = angularObjectRegistry.get(varName, noteId, paragraphId);
@@ -863,12 +926,12 @@ private void angularObjectUpdated(NotebookSocket conn, HashSet userAndRo
List settings = notebook.getInterpreterFactory()
.getInterpreterSettings(note.getId());
for (InterpreterSetting setting : settings) {
- if (setting.getInterpreterGroup(n.getId()) == null) {
+ if (setting.getInterpreterGroup(user, n.getId()) == null) {
continue;
}
- if (interpreterGroupId.equals(setting.getInterpreterGroup(n.getId()).getId())) {
+ if (interpreterGroupId.equals(setting.getInterpreterGroup(user, n.getId()).getId())) {
AngularObjectRegistry angularObjectRegistry = setting
- .getInterpreterGroup(n.getId()).getAngularObjectRegistry();
+ .getInterpreterGroup(user, n.getId()).getAngularObjectRegistry();
this.broadcastExcept(
n.getId(),
new Message(OP.ANGULAR_OBJECT_UPDATE).put("angularObject", ao)
@@ -1054,7 +1117,7 @@ private void moveParagraph(NotebookSocket conn, HashSet userAndRoles, No
String noteId = getOpenNoteId(conn);
final Note note = notebook.getNote(noteId);
NotebookAuthorization notebookAuthorization = notebook.getNotebookAuthorization();
- AuthenticationInfo subject = new AuthenticationInfo(SecurityUtils.getPrincipal());
+ AuthenticationInfo subject = new AuthenticationInfo(fromMessage.principal);
if (!notebookAuthorization.isWriter(noteId, userAndRoles)) {
permissionError(conn, "write", fromMessage.principal,
userAndRoles, notebookAuthorization.getWriters(noteId));
@@ -1073,7 +1136,7 @@ private void insertParagraph(NotebookSocket conn, HashSet userAndRoles,
String noteId = getOpenNoteId(conn);
final Note note = notebook.getNote(noteId);
NotebookAuthorization notebookAuthorization = notebook.getNotebookAuthorization();
- AuthenticationInfo subject = new AuthenticationInfo(SecurityUtils.getPrincipal());
+ AuthenticationInfo subject = new AuthenticationInfo(fromMessage.principal);
if (!notebookAuthorization.isWriter(noteId, userAndRoles)) {
permissionError(conn, "write", fromMessage.principal,
userAndRoles, notebookAuthorization.getWriters(noteId));
@@ -1125,14 +1188,9 @@ private void runParagraph(NotebookSocket conn, HashSet userAndRoles, Not
String text = (String) fromMessage.get("paragraph");
p.setText(text);
p.setTitle((String) fromMessage.get("title"));
- if (!fromMessage.principal.equals("anonymous")) {
- AuthenticationInfo authenticationInfo = new AuthenticationInfo(fromMessage.principal,
- fromMessage.ticket);
- p.setAuthenticationInfo(authenticationInfo);
-
- } else {
- p.setAuthenticationInfo(new AuthenticationInfo());
- }
+ AuthenticationInfo authenticationInfo =
+ new AuthenticationInfo(fromMessage.principal, fromMessage.ticket);
+ p.setAuthenticationInfo(authenticationInfo);
Map params = (Map) fromMessage
.get("params");
@@ -1457,7 +1515,7 @@ public void afterStatusChange(Job job, Status before, Status after) {
LOG.info("Job {} is finished", job.getId());
try {
//TODO(khalid): may change interface for JobListener and pass subject from interpreter
- note.persist(null);
+ note.persist(job instanceof Paragraph ? ((Paragraph) job).getAuthenticationInfo() : null);
} catch (IOException e) {
LOG.error(e.toString(), e);
}
@@ -1513,7 +1571,8 @@ public NotebookEventListener getNotebookInformationListener() {
return new NotebookInformationListener(this);
}
- private void sendAllAngularObjects(Note note, NotebookSocket conn) throws IOException {
+ private void sendAllAngularObjects(Note note, String user, NotebookSocket conn)
+ throws IOException {
List settings =
notebook().getInterpreterFactory().getInterpreterSettings(note.getId());
if (settings == null || settings.size() == 0) {
@@ -1521,17 +1580,15 @@ private void sendAllAngularObjects(Note note, NotebookSocket conn) throws IOExce
}
for (InterpreterSetting intpSetting : settings) {
- AngularObjectRegistry registry = intpSetting.getInterpreterGroup(note.getId())
- .getAngularObjectRegistry();
+ AngularObjectRegistry registry =
+ intpSetting.getInterpreterGroup(user, note.getId()).getAngularObjectRegistry();
List objects = registry.getAllWithGlobal(note.getId());
for (AngularObject object : objects) {
- conn.send(serializeMessage(new Message(OP.ANGULAR_OBJECT_UPDATE)
- .put("angularObject", object)
- .put("interpreterGroupId",
- intpSetting.getInterpreterGroup(note.getId()).getId())
- .put("noteId", note.getId())
- .put("paragraphId", object.getParagraphId())
- ));
+ conn.send(serializeMessage(
+ new Message(OP.ANGULAR_OBJECT_UPDATE).put("angularObject", object)
+ .put("interpreterGroupId",
+ intpSetting.getInterpreterGroup(user, note.getId()).getId())
+ .put("noteId", note.getId()).put("paragraphId", object.getParagraphId())));
}
}
}
@@ -1598,9 +1655,10 @@ private void getEditorSetting(NotebookSocket conn, Message fromMessage)
String paragraphId = (String) fromMessage.get("paragraphId");
String replName = (String) fromMessage.get("magic");
String noteId = getOpenNoteId(conn);
+ String user = fromMessage.principal;
Message resp = new Message(OP.EDITOR_SETTING);
resp.put("paragraphId", paragraphId);
- resp.put("editor", notebook().getInterpreterFactory().getEditorSetting(noteId, replName));
+ resp.put("editor", notebook().getInterpreterFactory().getEditorSetting(user, noteId, replName));
conn.send(serializeMessage(resp));
return;
}
diff --git a/zeppelin-server/src/main/java/org/apache/zeppelin/socket/NotebookSocket.java b/zeppelin-server/src/main/java/org/apache/zeppelin/socket/NotebookSocket.java
index f491ed708b6..baee746ed0f 100644
--- a/zeppelin-server/src/main/java/org/apache/zeppelin/socket/NotebookSocket.java
+++ b/zeppelin-server/src/main/java/org/apache/zeppelin/socket/NotebookSocket.java
@@ -20,6 +20,7 @@
import javax.servlet.http.HttpServletRequest;
+import org.apache.commons.lang.StringUtils;
import org.eclipse.jetty.websocket.api.Session;
import org.eclipse.jetty.websocket.api.WebSocketAdapter;
@@ -32,12 +33,14 @@ public class NotebookSocket extends WebSocketAdapter {
private NotebookSocketListener listener;
private HttpServletRequest request;
private String protocol;
+ private String user;
public NotebookSocket(HttpServletRequest req, String protocol,
NotebookSocketListener listener) {
this.listener = listener;
this.request = req;
this.protocol = protocol;
+ this.user = StringUtils.EMPTY;
}
@Override
@@ -69,4 +72,11 @@ public synchronized void send(String serializeMessage) throws IOException {
connection.getRemote().sendString(serializeMessage);
}
+ public String getUser() {
+ return user;
+ }
+
+ public void setUser(String user) {
+ this.user = user;
+ }
}
diff --git a/zeppelin-server/src/main/java/org/apache/zeppelin/utils/SecurityUtils.java b/zeppelin-server/src/main/java/org/apache/zeppelin/utils/SecurityUtils.java
index f9e5929a882..d81d2e6150b 100644
--- a/zeppelin-server/src/main/java/org/apache/zeppelin/utils/SecurityUtils.java
+++ b/zeppelin-server/src/main/java/org/apache/zeppelin/utils/SecurityUtils.java
@@ -119,4 +119,10 @@ public static HashSet getRoles() {
return roles;
}
+ /**
+ * Checked if shiro enabled or not
+ */
+ public static boolean isAuthenticated() {
+ return org.apache.shiro.SecurityUtils.getSubject().isAuthenticated();
+ }
}
diff --git a/zeppelin-server/src/main/resources/log4j.properties b/zeppelin-server/src/main/resources/log4j.properties
new file mode 100644
index 00000000000..2f644074c89
--- /dev/null
+++ b/zeppelin-server/src/main/resources/log4j.properties
@@ -0,0 +1,25 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+log4j.rootLogger = INFO, stdout
+
+log4j.appender.stdout = org.apache.log4j.ConsoleAppender
+log4j.appender.stdout.layout = org.apache.log4j.PatternLayout
+log4j.appender.stdout.layout.ConversionPattern=%5p [%d] ({%t} %F[%M]:%L) - %m%n
+
+log4j.additivity.org.apache.zeppelin.interpreter = false
+log4j.logger.org.apache.zeppelin.interpreter = DEBUG, stdout
diff --git a/zeppelin-server/src/main/resources/shiro.ini b/zeppelin-server/src/main/resources/shiro.ini
index 371a44e11e1..050c9d99bb5 100644
--- a/zeppelin-server/src/main/resources/shiro.ini
+++ b/zeppelin-server/src/main/resources/shiro.ini
@@ -18,14 +18,19 @@
[users]
# List of users with their password allowed to access Zeppelin.
# To use a different strategy (LDAP / Database / ...) check the shiro doc at http://shiro.apache.org/configuration.html#Configuration-INISections
-admin = password
+admin = password, admin
+user1 = user1, role1
[urls]
-
# anon means the access is anonymous.
# authcBasic means Basic Auth Security
# To enfore security, comment the line below and uncomment the next one
/** = anon
-#/** = authcBasic
+#/** = authc
+[roles]
+role1 = *
+role2 = *
+role3 = *
+admin = *
diff --git a/zeppelin-server/src/test/java/org/apache/zeppelin/integration/AuthenticationIT.java b/zeppelin-server/src/test/java/org/apache/zeppelin/integration/AuthenticationIT.java
index 60014298566..e1a6f9efb01 100644
--- a/zeppelin-server/src/test/java/org/apache/zeppelin/integration/AuthenticationIT.java
+++ b/zeppelin-server/src/test/java/org/apache/zeppelin/integration/AuthenticationIT.java
@@ -202,7 +202,7 @@ public void testGroupPermission() throws Exception {
try {
WebElement element = pollingWait(By.xpath("//*[@id='notebook-names']//a[contains(@href, '" + noteId + "')]"),
MAX_BROWSER_TIMEOUT_SEC);
- collector.checkThat("Check is user has permission to view this notebook link", false,
+ collector.checkThat("Check is user has permission to view this notebook link", true,
CoreMatchers.equalTo(element.isDisplayed()));
} catch (Exception e) {
//This should have failed, nothing to worry.
diff --git a/zeppelin-server/src/test/java/org/apache/zeppelin/rest/InterpreterRestApiTest.java b/zeppelin-server/src/test/java/org/apache/zeppelin/rest/InterpreterRestApiTest.java
index c767eb05786..6d4fb2c5472 100644
--- a/zeppelin-server/src/test/java/org/apache/zeppelin/rest/InterpreterRestApiTest.java
+++ b/zeppelin-server/src/test/java/org/apache/zeppelin/rest/InterpreterRestApiTest.java
@@ -30,7 +30,9 @@
import org.apache.zeppelin.notebook.Paragraph;
import org.apache.zeppelin.scheduler.Job.Status;
import org.apache.zeppelin.server.ZeppelinServer;
+import org.apache.zeppelin.user.AuthenticationInfo;
import org.junit.AfterClass;
+import org.junit.Before;
import org.junit.BeforeClass;
import org.junit.FixMethodOrder;
import org.junit.Test;
@@ -47,6 +49,7 @@
@FixMethodOrder(MethodSorters.NAME_ASCENDING)
public class InterpreterRestApiTest extends AbstractTestRestApi {
Gson gson = new Gson();
+ AuthenticationInfo anonymous;
@BeforeClass
public static void init() throws Exception {
@@ -58,6 +61,11 @@ public static void destroy() throws Exception {
AbstractTestRestApi.shutDown();
}
+ @Before
+ public void setUp() {
+ anonymous = new AuthenticationInfo("anonymous");
+ }
+
@Test
public void getAvailableInterpreters() throws IOException {
// when
@@ -90,7 +98,7 @@ public void testSettingsCRUD() throws IOException {
String jsonRequest = "{\"name\":\"md2\",\"group\":\"md\",\"properties\":{\"propname\":\"propvalue\"}," +
"\"interpreterGroup\":[{\"class\":\"org.apache.zeppelin.markdown.Markdown\",\"name\":\"md\"}]," +
"\"dependencies\":[]," +
- "\"option\": { \"remote\": true, \"perNoteSession\": false }}";
+ "\"option\": { \"remote\": true, \"session\": false }}";
PostMethod post = httpPost("/interpreter/setting/", jsonRequest);
LOG.info("testSettingCRUD create response\n" + post.getResponseBodyAsString());
assertThat("test create method:", post, isCreated());
@@ -106,7 +114,7 @@ public void testSettingsCRUD() throws IOException {
jsonRequest = "{\"name\":\"md2\",\"group\":\"md\",\"properties\":{\"propname\":\"Otherpropvalue\"}," +
"\"interpreterGroup\":[{\"class\":\"org.apache.zeppelin.markdown.Markdown\",\"name\":\"md\"}]," +
"\"dependencies\":[]," +
- "\"option\": { \"remote\": true, \"perNoteSession\": false }}";
+ "\"option\": { \"remote\": true, \"session\": false }}";
PutMethod put = httpPut("/interpreter/setting/" + newSettingId, jsonRequest);
LOG.info("testSettingCRUD update response\n" + put.getResponseBodyAsString());
assertThat("test update method:", put, isAllowed());
@@ -131,7 +139,7 @@ public void testSettingsCreateWithEmptyJson() throws IOException {
@Test
public void testInterpreterAutoBinding() throws IOException {
// create note
- Note note = ZeppelinServer.notebook.createNote(null);
+ Note note = ZeppelinServer.notebook.createNote(anonymous);
// check interpreter is binded
GetMethod get = httpGet("/notebook/interpreter/bind/" + note.getId());
@@ -144,13 +152,13 @@ public void testInterpreterAutoBinding() throws IOException {
get.releaseConnection();
//cleanup
- ZeppelinServer.notebook.removeNote(note.getId(), null);
+ ZeppelinServer.notebook.removeNote(note.getId(), anonymous);
}
@Test
public void testInterpreterRestart() throws IOException, InterruptedException {
// create new note
- Note note = ZeppelinServer.notebook.createNote(null);
+ Note note = ZeppelinServer.notebook.createNote(anonymous);
note.addParagraph();
Paragraph p = note.getLastParagraph();
Map config = p.getConfig();
@@ -159,6 +167,7 @@ public void testInterpreterRestart() throws IOException, InterruptedException {
// run markdown paragraph
p.setConfig(config);
p.setText("%md markdown");
+ p.setAuthenticationInfo(anonymous);
note.run(p.getId());
while (p.getStatus() != Status.FINISHED) {
Thread.sleep(100);
@@ -181,13 +190,14 @@ public void testInterpreterRestart() throws IOException, InterruptedException {
p = note.addParagraph();
p.setConfig(config);
p.setText("%md markdown restarted");
+ p.setAuthenticationInfo(anonymous);
note.run(p.getId());
while (p.getStatus() != Status.FINISHED) {
Thread.sleep(100);
}
assertEquals("markdown restarted
\n", p.getResult().message());
//cleanup
- ZeppelinServer.notebook.removeNote(note.getId(), null);
+ ZeppelinServer.notebook.removeNote(note.getId(), anonymous);
}
@Test
diff --git a/zeppelin-server/src/test/java/org/apache/zeppelin/rest/NotebookRestApiTest.java b/zeppelin-server/src/test/java/org/apache/zeppelin/rest/NotebookRestApiTest.java
index d7f55f54744..9383569fae1 100644
--- a/zeppelin-server/src/test/java/org/apache/zeppelin/rest/NotebookRestApiTest.java
+++ b/zeppelin-server/src/test/java/org/apache/zeppelin/rest/NotebookRestApiTest.java
@@ -28,7 +28,9 @@
import org.apache.zeppelin.notebook.NotebookAuthorization;
import org.apache.zeppelin.notebook.NotebookAuthorizationInfoSaving;
import org.apache.zeppelin.server.ZeppelinServer;
+import org.apache.zeppelin.user.AuthenticationInfo;
import org.junit.AfterClass;
+import org.junit.Before;
import org.junit.BeforeClass;
import org.junit.FixMethodOrder;
import org.junit.Test;
@@ -49,6 +51,7 @@
@FixMethodOrder(MethodSorters.NAME_ASCENDING)
public class NotebookRestApiTest extends AbstractTestRestApi {
Gson gson = new Gson();
+ AuthenticationInfo anonymous;
@BeforeClass
public static void init() throws Exception {
@@ -60,9 +63,14 @@ public static void destroy() throws Exception {
AbstractTestRestApi.shutDown();
}
+ @Before
+ public void setUp() {
+ anonymous = new AuthenticationInfo("anonymous");
+ }
+
@Test
public void testPermissions() throws IOException {
- Note note1 = ZeppelinServer.notebook.createNote(null);
+ Note note1 = ZeppelinServer.notebook.createNote(anonymous);
// Set only readers
String jsonRequest = "{\"readers\":[\"admin-team\"],\"owners\":[]," +
"\"writers\":[]}";
@@ -85,7 +93,7 @@ public void testPermissions() throws IOException {
get.releaseConnection();
- Note note2 = ZeppelinServer.notebook.createNote(null);
+ Note note2 = ZeppelinServer.notebook.createNote(anonymous);
// Set only writers
jsonRequest = "{\"readers\":[],\"owners\":[]," +
"\"writers\":[\"admin-team\"]}";
@@ -119,14 +127,14 @@ public void testPermissions() throws IOException {
assertEquals(authInfo.get("owners"), Lists.newArrayList());
get.releaseConnection();
//cleanup
- ZeppelinServer.notebook.removeNote(note1.getId(), null);
- ZeppelinServer.notebook.removeNote(note2.getId(), null);
+ ZeppelinServer.notebook.removeNote(note1.getId(), anonymous);
+ ZeppelinServer.notebook.removeNote(note2.getId(), anonymous);
}
@Test
public void testGetNoteParagraphJobStatus() throws IOException {
- Note note1 = ZeppelinServer.notebook.createNote(null);
+ Note note1 = ZeppelinServer.notebook.createNote(anonymous);
note1.addParagraph();
String paragraphId = note1.getLastParagraph().getId();
@@ -142,13 +150,13 @@ public void testGetNoteParagraphJobStatus() throws IOException {
assertEquals(paragraphStatus.get("status"), "READY");
//cleanup
- ZeppelinServer.notebook.removeNote(note1.getId(), null);
+ ZeppelinServer.notebook.removeNote(note1.getId(), anonymous);
}
@Test
public void testCloneNotebook() throws IOException {
- Note note1 = ZeppelinServer.notebook.createNote(null);
+ Note note1 = ZeppelinServer.notebook.createNote(anonymous);
PostMethod post = httpPost("/notebook/" + note1.getId(), "");
LOG.info("testCloneNotebook response\n" + post.getResponseBodyAsString());
assertThat(post, isCreated());
@@ -167,8 +175,8 @@ public void testCloneNotebook() throws IOException {
get.releaseConnection();
//cleanup
- ZeppelinServer.notebook.removeNote(note1.getId(), null);
- ZeppelinServer.notebook.removeNote(clonedNotebookId, null);
+ ZeppelinServer.notebook.removeNote(note1.getId(), anonymous);
+ ZeppelinServer.notebook.removeNote(clonedNotebookId, anonymous);
}
}
diff --git a/zeppelin-server/src/test/java/org/apache/zeppelin/rest/ZeppelinRestApiTest.java b/zeppelin-server/src/test/java/org/apache/zeppelin/rest/ZeppelinRestApiTest.java
index 4390d74b495..3b92c14550c 100644
--- a/zeppelin-server/src/test/java/org/apache/zeppelin/rest/ZeppelinRestApiTest.java
+++ b/zeppelin-server/src/test/java/org/apache/zeppelin/rest/ZeppelinRestApiTest.java
@@ -19,9 +19,11 @@
import java.io.IOException;
import java.util.ArrayList;
+import java.util.HashSet;
import java.util.List;
import java.util.Map;
+import com.google.common.collect.Sets;
import org.apache.commons.httpclient.methods.DeleteMethod;
import org.apache.commons.httpclient.methods.GetMethod;
import org.apache.commons.httpclient.methods.PostMethod;
@@ -32,7 +34,9 @@
import org.apache.zeppelin.notebook.Paragraph;
import org.apache.zeppelin.scheduler.Job.Status;
import org.apache.zeppelin.server.ZeppelinServer;
+import org.apache.zeppelin.user.AuthenticationInfo;
import org.junit.AfterClass;
+import org.junit.Before;
import org.junit.BeforeClass;
import org.junit.FixMethodOrder;
import org.junit.Test;
@@ -50,6 +54,7 @@
@FixMethodOrder(MethodSorters.NAME_ASCENDING)
public class ZeppelinRestApiTest extends AbstractTestRestApi {
Gson gson = new Gson();
+ AuthenticationInfo anonymous;
@BeforeClass
public static void init() throws Exception {
@@ -61,6 +66,11 @@ public static void destroy() throws Exception {
AbstractTestRestApi.shutDown();
}
+ @Before
+ public void setUp() {
+ anonymous = new AuthenticationInfo("anonymous");
+ }
+
/***
* ROOT API TEST
***/
@@ -77,7 +87,7 @@ public void getApiRoot() throws IOException {
public void testGetNotebookInfo() throws IOException {
LOG.info("testGetNotebookInfo");
// Create note to get info
- Note note = ZeppelinServer.notebook.createNote(null);
+ Note note = ZeppelinServer.notebook.createNote(anonymous);
assertNotNull("can't create new note", note);
note.setName("note");
Paragraph paragraph = note.addParagraph();
@@ -86,7 +96,7 @@ public void testGetNotebookInfo() throws IOException {
paragraph.setConfig(config);
String paragraphText = "%md This is my new paragraph in my new note";
paragraph.setText(paragraphText);
- note.persist(null);
+ note.persist(anonymous);
String sourceNoteID = note.getId();
GetMethod get = httpGet("/notebook/" + sourceNoteID);
@@ -104,6 +114,8 @@ public void testGetNotebookInfo() throws IOException {
assertTrue(paragraphs.size() > 0);
assertEquals(paragraphText, paragraphs.get(0).get("text"));
+ //
+ ZeppelinServer.notebook.removeNote(sourceNoteID, anonymous);
}
@Test
@@ -153,7 +165,7 @@ public void testNotebookCreateWithParagraphs() throws IOException {
assertTrue("paragraph text check failed", p.getText().startsWith("text"));
}
// cleanup
- ZeppelinServer.notebook.removeNote(newNotebookId, null);
+ ZeppelinServer.notebook.removeNote(newNotebookId, anonymous);
post.releaseConnection();
}
@@ -180,7 +192,7 @@ private void testNotebookCreate(String noteName) throws IOException {
}
assertEquals("compare note name", expectedNoteName, newNoteName);
// cleanup
- ZeppelinServer.notebook.removeNote(newNotebookId, null);
+ ZeppelinServer.notebook.removeNote(newNotebookId, anonymous);
post.releaseConnection();
}
@@ -189,7 +201,7 @@ private void testNotebookCreate(String noteName) throws IOException {
public void testDeleteNote() throws IOException {
LOG.info("testDeleteNote");
//Create note and get ID
- Note note = ZeppelinServer.notebook.createNote(null);
+ Note note = ZeppelinServer.notebook.createNote(anonymous);
String noteId = note.getId();
testDeleteNotebook(noteId);
}
@@ -205,7 +217,7 @@ public void testDeleteNoteBadId() throws IOException {
@Test
public void testExportNotebook() throws IOException {
LOG.info("testExportNotebook");
- Note note = ZeppelinServer.notebook.createNote(null);
+ Note note = ZeppelinServer.notebook.createNote(anonymous);
assertNotNull("can't create new note", note);
note.setName("source note for export");
Paragraph paragraph = note.addParagraph();
@@ -213,7 +225,7 @@ public void testExportNotebook() throws IOException {
config.put("enabled", true);
paragraph.setConfig(config);
paragraph.setText("%md This is my new paragraph in my new note");
- note.persist(null);
+ note.persist(anonymous);
String sourceNoteID = note.getId();
// Call export Notebook REST API
GetMethod get = httpGet("/notebook/export/" + sourceNoteID);
@@ -227,7 +239,7 @@ public void testExportNotebook() throws IOException {
String exportJSON = (String) resp.get("body");
assertNotNull("Can not find new notejson", exportJSON);
LOG.info("export JSON:=" + exportJSON);
- ZeppelinServer.notebook.removeNote(sourceNoteID, null);
+ ZeppelinServer.notebook.removeNote(sourceNoteID, anonymous);
get.releaseConnection();
}
@@ -238,7 +250,7 @@ public void testImportNotebook() throws IOException {
String noteName = "source note for import";
LOG.info("testImortNotebook");
// create test notebook
- Note note = ZeppelinServer.notebook.createNote(null);
+ Note note = ZeppelinServer.notebook.createNote(anonymous);
assertNotNull("can't create new note", note);
note.setName(noteName);
Paragraph paragraph = note.addParagraph();
@@ -246,7 +258,7 @@ public void testImportNotebook() throws IOException {
config.put("enabled", true);
paragraph.setConfig(config);
paragraph.setText("%md This is my new paragraph in my new note");
- note.persist(null);
+ note.persist(anonymous);
String sourceNoteID = note.getId();
// get note content as JSON
String oldJson = getNoteContent(sourceNoteID);
@@ -264,8 +276,8 @@ public void testImportNotebook() throws IOException {
assertEquals("Compare paragraphs count", note.getParagraphs().size(), newNote.getParagraphs()
.size());
// cleanup
- ZeppelinServer.notebook.removeNote(note.getId(), null);
- ZeppelinServer.notebook.removeNote(newNote.getId(), null);
+ ZeppelinServer.notebook.removeNote(note.getId(), anonymous);
+ ZeppelinServer.notebook.removeNote(newNote.getId(), anonymous);
importPost.releaseConnection();
}
@@ -300,7 +312,7 @@ private void testDeleteNotebook(String notebookId) throws IOException {
public void testCloneNotebook() throws IOException, CloneNotSupportedException, IllegalArgumentException {
LOG.info("testCloneNotebook");
// Create note to clone
- Note note = ZeppelinServer.notebook.createNote(null);
+ Note note = ZeppelinServer.notebook.createNote(anonymous);
assertNotNull("can't create new note", note);
note.setName("source note for clone");
Paragraph paragraph = note.addParagraph();
@@ -308,7 +320,7 @@ public void testCloneNotebook() throws IOException, CloneNotSupportedException,
config.put("enabled", true);
paragraph.setConfig(config);
paragraph.setText("%md This is my new paragraph in my new note");
- note.persist(null);
+ note.persist(anonymous);
String sourceNoteID = note.getId();
String noteName = "clone Note Name";
@@ -328,8 +340,8 @@ public void testCloneNotebook() throws IOException, CloneNotSupportedException,
assertEquals("Compare note names", noteName, newNote.getName());
assertEquals("Compare paragraphs count", note.getParagraphs().size(), newNote.getParagraphs().size());
//cleanup
- ZeppelinServer.notebook.removeNote(note.getId(), null);
- ZeppelinServer.notebook.removeNote(newNote.getId(), null);
+ ZeppelinServer.notebook.removeNote(note.getId(), anonymous);
+ ZeppelinServer.notebook.removeNote(newNote.getId(), anonymous);
post.releaseConnection();
}
@@ -341,7 +353,9 @@ public void testListNotebooks() throws IOException {
Map resp = gson.fromJson(get.getResponseBodyAsString(), new TypeToken>() {
}.getType());
List> body = (List>) resp.get("body");
- assertEquals("List notebooks are equal", ZeppelinServer.notebook.getAllNotes().size(), body.size());
+ //TODO(khalid): anonymous or specific user notes?
+ HashSet anonymous = Sets.newHashSet("anonymous");
+ assertEquals("List notebooks are equal", ZeppelinServer.notebook.getAllNotes(anonymous).size(), body.size());
get.releaseConnection();
}
@@ -349,7 +363,7 @@ public void testListNotebooks() throws IOException {
public void testNoteJobs() throws IOException, InterruptedException {
LOG.info("testNoteJobs");
// Create note to run test.
- Note note = ZeppelinServer.notebook.createNote(null);
+ Note note = ZeppelinServer.notebook.createNote(anonymous);
assertNotNull("can't create new note", note);
note.setName("note for run test");
Paragraph paragraph = note.addParagraph();
@@ -359,7 +373,7 @@ public void testNoteJobs() throws IOException, InterruptedException {
paragraph.setConfig(config);
paragraph.setText("%md This is test paragraph.");
- note.persist(null);
+ note.persist(anonymous);
String noteID = note.getId();
note.runAll();
@@ -397,14 +411,14 @@ public void testNoteJobs() throws IOException, InterruptedException {
Thread.sleep(1000);
//cleanup
- ZeppelinServer.notebook.removeNote(note.getId(), null);
+ ZeppelinServer.notebook.removeNote(note.getId(), anonymous);
}
@Test
public void testGetNotebookJob() throws IOException, InterruptedException {
LOG.info("testGetNotebookJob");
// Create note to run test.
- Note note = ZeppelinServer.notebook.createNote(null);
+ Note note = ZeppelinServer.notebook.createNote(anonymous);
assertNotNull("can't create new note", note);
note.setName("note for run test");
Paragraph paragraph = note.addParagraph();
@@ -414,7 +428,8 @@ public void testGetNotebookJob() throws IOException, InterruptedException {
paragraph.setConfig(config);
paragraph.setText("%sh sleep 1");
- note.persist(null);
+ paragraph.setAuthenticationInfo(anonymous);
+ note.persist(anonymous);
String noteID = note.getId();
note.runAll();
@@ -450,14 +465,14 @@ public void testGetNotebookJob() throws IOException, InterruptedException {
}
}
- ZeppelinServer.notebook.removeNote(note.getId(), null);
+ ZeppelinServer.notebook.removeNote(note.getId(), anonymous);
}
@Test
public void testRunParagraphWithParams() throws IOException, InterruptedException {
LOG.info("testRunParagraphWithParams");
// Create note to run test.
- Note note = ZeppelinServer.notebook.createNote(null);
+ Note note = ZeppelinServer.notebook.createNote(anonymous);
assertNotNull("can't create new note", note);
note.setName("note for run test");
Paragraph paragraph = note.addParagraph();
@@ -467,7 +482,7 @@ public void testRunParagraphWithParams() throws IOException, InterruptedExceptio
paragraph.setConfig(config);
paragraph.setText("%spark\nval param = z.input(\"param\").toString\nprintln(param)");
- note.persist(null);
+ note.persist(anonymous);
String noteID = note.getId();
note.runAll();
@@ -495,13 +510,13 @@ public void testRunParagraphWithParams() throws IOException, InterruptedExceptio
assertEquals("world", params.get("param2"));
//cleanup
- ZeppelinServer.notebook.removeNote(note.getId(), null);
+ ZeppelinServer.notebook.removeNote(note.getId(), anonymous);
}
@Test
- public void testCronJobs() throws InterruptedException, IOException{
+ public void testJobs() throws InterruptedException, IOException{
// create a note and a paragraph
- Note note = ZeppelinServer.notebook.createNote(null);
+ Note note = ZeppelinServer.notebook.createNote(anonymous);
note.setName("note for run test");
Paragraph paragraph = note.addParagraph();
@@ -545,18 +560,18 @@ public void testCronJobs() throws InterruptedException, IOException{
DeleteMethod deleteCron = httpDelete("/notebook/cron/" + note.getId());
assertThat("", deleteCron, isAllowed());
deleteCron.releaseConnection();
- ZeppelinServer.notebook.removeNote(note.getId(), null);
+ ZeppelinServer.notebook.removeNote(note.getId(), anonymous);
}
@Test
public void testRegressionZEPPELIN_527() throws IOException {
- Note note = ZeppelinServer.notebook.createNote(null);
+ Note note = ZeppelinServer.notebook.createNote(anonymous);
note.setName("note for run test");
Paragraph paragraph = note.addParagraph();
paragraph.setText("%spark\nval param = z.input(\"param\").toString\nprintln(param)");
- note.persist(null);
+ note.persist(anonymous);
GetMethod getNoteJobs = httpGet("/notebook/job/" + note.getId());
assertThat("test notebook jobs run:", getNoteJobs, isAllowed());
@@ -567,12 +582,12 @@ public void testRegressionZEPPELIN_527() throws IOException {
assertFalse(body.get(0).containsKey("finished"));
getNoteJobs.releaseConnection();
- ZeppelinServer.notebook.removeNote(note.getId(), null);
+ ZeppelinServer.notebook.removeNote(note.getId(), anonymous);
}
@Test
public void testInsertParagraph() throws IOException {
- Note note = ZeppelinServer.notebook.createNote(null);
+ Note note = ZeppelinServer.notebook.createNote(anonymous);
String jsonRequest = "{\"title\": \"title1\", \"text\": \"text1\"}";
PostMethod post = httpPost("/notebook/" + note.getId() + "/paragraph", jsonRequest);
@@ -607,17 +622,17 @@ public void testInsertParagraph() throws IOException {
assertEquals("title2", paragraphAtIdx0.getTitle());
assertEquals("text2", paragraphAtIdx0.getText());
- ZeppelinServer.notebook.removeNote(note.getId(), null);
+ ZeppelinServer.notebook.removeNote(note.getId(), anonymous);
}
@Test
public void testGetParagraph() throws IOException {
- Note note = ZeppelinServer.notebook.createNote(null);
+ Note note = ZeppelinServer.notebook.createNote(anonymous);
Paragraph p = note.addParagraph();
p.setTitle("hello");
p.setText("world");
- note.persist(null);
+ note.persist(anonymous);
GetMethod get = httpGet("/notebook/" + note.getId() + "/paragraph/" + p.getId());
LOG.info("testGetParagraph response\n" + get.getResponseBodyAsString());
@@ -636,12 +651,12 @@ public void testGetParagraph() throws IOException {
assertEquals("hello", body.get("title"));
assertEquals("world", body.get("text"));
- ZeppelinServer.notebook.removeNote(note.getId(), null);
+ ZeppelinServer.notebook.removeNote(note.getId(), anonymous);
}
@Test
public void testMoveParagraph() throws IOException {
- Note note = ZeppelinServer.notebook.createNote(null);
+ Note note = ZeppelinServer.notebook.createNote(anonymous);
Paragraph p = note.addParagraph();
p.setTitle("title1");
@@ -651,7 +666,7 @@ public void testMoveParagraph() throws IOException {
p2.setTitle("title2");
p2.setText("text2");
- note.persist(null);
+ note.persist(anonymous);
PostMethod post = httpPost("/notebook/" + note.getId() + "/paragraph/" + p2.getId() + "/move/" + 0, "");
assertThat("Test post method: ", post, isAllowed());
@@ -668,18 +683,18 @@ public void testMoveParagraph() throws IOException {
assertThat("Test post method: ", post2, isBadRequest());
post.releaseConnection();
- ZeppelinServer.notebook.removeNote(note.getId(), null);
+ ZeppelinServer.notebook.removeNote(note.getId(), anonymous);
}
@Test
public void testDeleteParagraph() throws IOException {
- Note note = ZeppelinServer.notebook.createNote(null);
+ Note note = ZeppelinServer.notebook.createNote(anonymous);
Paragraph p = note.addParagraph();
p.setTitle("title1");
p.setText("text1");
- note.persist(null);
+ note.persist(anonymous);
DeleteMethod delete = httpDelete("/notebook/" + note.getId() + "/paragraph/" + p.getId());
assertThat("Test delete method: ", delete, isAllowed());
@@ -689,7 +704,7 @@ public void testDeleteParagraph() throws IOException {
Paragraph retrParagrah = retrNote.getParagraph(p.getId());
assertNull("paragraph should be deleted", retrParagrah);
- ZeppelinServer.notebook.removeNote(note.getId(), null);
+ ZeppelinServer.notebook.removeNote(note.getId(), anonymous);
}
@Test
@@ -705,12 +720,12 @@ public void testSearch() throws IOException {
String username = body.get("principal");
getSecurityTicket.releaseConnection();
- Note note1 = ZeppelinServer.notebook.createNote(null);
+ Note note1 = ZeppelinServer.notebook.createNote(anonymous);
String jsonRequest = "{\"title\": \"title1\", \"text\": \"ThisIsToTestSearchMethodWithPermissions 1\"}";
PostMethod postNotebookText = httpPost("/notebook/" + note1.getId() + "/paragraph", jsonRequest);
postNotebookText.releaseConnection();
- Note note2 = ZeppelinServer.notebook.createNote(null);
+ Note note2 = ZeppelinServer.notebook.createNote(anonymous);
jsonRequest = "{\"title\": \"title1\", \"text\": \"ThisIsToTestSearchMethodWithPermissions 2\"}";
postNotebookText = httpPost("/notebook/" + note2.getId() + "/paragraph", jsonRequest);
postNotebookText.releaseConnection();
@@ -752,13 +767,13 @@ public void testSearch() throws IOException {
getPermission.releaseConnection();
}
searchNotebook.releaseConnection();
- ZeppelinServer.notebook.removeNote(note1.getId(), null);
- ZeppelinServer.notebook.removeNote(note2.getId(), null);
+ ZeppelinServer.notebook.removeNote(note1.getId(), anonymous);
+ ZeppelinServer.notebook.removeNote(note2.getId(), anonymous);
}
@Test
public void testTitleSearch() throws IOException {
- Note note = ZeppelinServer.notebook.createNote(null);
+ Note note = ZeppelinServer.notebook.createNote(anonymous);
String jsonRequest = "{\"title\": \"testTitleSearchOfParagraph\", \"text\": \"ThisIsToTestSearchMethodWithTitle \"}";
PostMethod postNotebookText = httpPost("/notebook/" + note.getId() + "/paragraph", jsonRequest);
postNotebookText.releaseConnection();
@@ -779,7 +794,7 @@ public void testTitleSearch() throws IOException {
}
assertEquals("Paragraph title hits must be at-least one", true, numberOfTitleHits >= 1);
searchNotebook.releaseConnection();
- ZeppelinServer.notebook.removeNote(note.getId(), null);
+ ZeppelinServer.notebook.removeNote(note.getId(), anonymous);
}
}
diff --git a/zeppelin-server/src/test/java/org/apache/zeppelin/rest/ZeppelinSparkClusterTest.java b/zeppelin-server/src/test/java/org/apache/zeppelin/rest/ZeppelinSparkClusterTest.java
index 5084ae73719..740ef4066c4 100644
--- a/zeppelin-server/src/test/java/org/apache/zeppelin/rest/ZeppelinSparkClusterTest.java
+++ b/zeppelin-server/src/test/java/org/apache/zeppelin/rest/ZeppelinSparkClusterTest.java
@@ -31,7 +31,9 @@
import org.apache.zeppelin.notebook.Paragraph;
import org.apache.zeppelin.scheduler.Job.Status;
import org.apache.zeppelin.server.ZeppelinServer;
+import org.apache.zeppelin.user.AuthenticationInfo;
import org.junit.AfterClass;
+import org.junit.Before;
import org.junit.BeforeClass;
import org.junit.Test;
@@ -43,6 +45,7 @@
*/
public class ZeppelinSparkClusterTest extends AbstractTestRestApi {
Gson gson = new Gson();
+ AuthenticationInfo anonymous;
@BeforeClass
public static void init() throws Exception {
@@ -54,6 +57,11 @@ public static void destroy() throws Exception {
AbstractTestRestApi.shutDown();
}
+ @Before
+ public void setUp() {
+ anonymous = new AuthenticationInfo("anonymous");
+ }
+
private void waitForFinish(Paragraph p) {
while (p.getStatus() != Status.FINISHED
&& p.getStatus() != Status.ERROR
@@ -69,7 +77,7 @@ private void waitForFinish(Paragraph p) {
@Test
public void basicRDDTransformationAndActionTest() throws IOException {
// create new note
- Note note = ZeppelinServer.notebook.createNote(null);
+ Note note = ZeppelinServer.notebook.createNote(anonymous);
// run markdown paragraph, again
Paragraph p = note.addParagraph();
@@ -77,17 +85,18 @@ public void basicRDDTransformationAndActionTest() throws IOException {
config.put("enabled", true);
p.setConfig(config);
p.setText("%spark print(sc.parallelize(1 to 10).reduce(_ + _))");
+ p.setAuthenticationInfo(anonymous);
note.run(p.getId());
waitForFinish(p);
assertEquals(Status.FINISHED, p.getStatus());
assertEquals("55", p.getResult().message());
- ZeppelinServer.notebook.removeNote(note.getId(), null);
+ ZeppelinServer.notebook.removeNote(note.getId(), anonymous);
}
@Test
public void sparkSQLTest() throws IOException {
// create new note
- Note note = ZeppelinServer.notebook.createNote(null);
+ Note note = ZeppelinServer.notebook.createNote(anonymous);
int sparkVersion = getSparkVersionNumber(note);
// DataFrame API is available from spark 1.3
if (sparkVersion >= 13) {
@@ -98,6 +107,7 @@ public void sparkSQLTest() throws IOException {
p.setConfig(config);
p.setText("%spark val df=sqlContext.createDataFrame(Seq((\"hello\",20)))\n" +
"df.collect()");
+ p.setAuthenticationInfo(anonymous);
note.run(p.getId());
waitForFinish(p);
assertEquals(Status.FINISHED, p.getStatus());
@@ -111,6 +121,7 @@ public void sparkSQLTest() throws IOException {
p.setConfig(config);
p.setText("%spark val df=sqlContext.createDataFrame(Seq((\"hello\",20)))\n" +
"z.show(df)");
+ p.setAuthenticationInfo(anonymous);
note.run(p.getId());
waitForFinish(p);
assertEquals(Status.FINISHED, p.getStatus());
@@ -125,20 +136,21 @@ public void sparkSQLTest() throws IOException {
p.setConfig(config);
p.setText("%spark val ds=spark.createDataset(Seq((\"hello\",20)))\n" +
"z.show(ds)");
+ p.setAuthenticationInfo(anonymous);
note.run(p.getId());
waitForFinish(p);
assertEquals(Status.FINISHED, p.getStatus());
assertEquals(InterpreterResult.Type.TABLE, p.getResult().type());
assertEquals("_1\t_2\nhello\t20\n", p.getResult().message());
}
- ZeppelinServer.notebook.removeNote(note.getId(), null);
+ ZeppelinServer.notebook.removeNote(note.getId(), anonymous);
}
}
@Test
public void sparkRTest() throws IOException {
// create new note
- Note note = ZeppelinServer.notebook.createNote(null);
+ Note note = ZeppelinServer.notebook.createNote(anonymous);
int sparkVersion = getSparkVersionNumber(note);
if (isSparkR() && sparkVersion >= 14) { // sparkr supported from 1.4.0
@@ -165,19 +177,20 @@ public void sparkRTest() throws IOException {
"df <- createDataFrame(" + sqlContextName + ", localDF)\n" +
"count(df)"
);
+ p.setAuthenticationInfo(anonymous);
note.run(p.getId());
waitForFinish(p);
System.err.println("sparkRTest=" + p.getResult().message());
assertEquals(Status.FINISHED, p.getStatus());
assertEquals("[1] 3", p.getResult().message());
}
- ZeppelinServer.notebook.removeNote(note.getId(), null);
+ ZeppelinServer.notebook.removeNote(note.getId(), anonymous);
}
@Test
public void pySparkTest() throws IOException {
// create new note
- Note note = ZeppelinServer.notebook.createNote(null);
+ Note note = ZeppelinServer.notebook.createNote(anonymous);
note.setName("note");
int sparkVersion = getSparkVersionNumber(note);
@@ -188,6 +201,7 @@ public void pySparkTest() throws IOException {
config.put("enabled", true);
p.setConfig(config);
p.setText("%pyspark print(sc.parallelize(range(1, 11)).reduce(lambda a, b: a + b))");
+ p.setAuthenticationInfo(anonymous);
note.run(p.getId());
waitForFinish(p);
assertEquals(Status.FINISHED, p.getStatus());
@@ -201,6 +215,7 @@ public void pySparkTest() throws IOException {
p.setText("%pyspark from pyspark.sql import Row\n" +
"df=sqlContext.createDataFrame([Row(id=1, age=20)])\n" +
"df.collect()");
+ p.setAuthenticationInfo(anonymous);
note.run(p.getId());
waitForFinish(p);
assertEquals(Status.FINISHED, p.getStatus());
@@ -214,6 +229,7 @@ public void pySparkTest() throws IOException {
p.setText("%pyspark from pyspark.sql import Row\n" +
"df=sqlContext.createDataFrame([Row(id=1, age=20)])\n" +
"z.show(df)");
+ p.setAuthenticationInfo(anonymous);
note.run(p.getId());
waitForFinish(p);
assertEquals(Status.FINISHED, p.getStatus());
@@ -228,6 +244,7 @@ public void pySparkTest() throws IOException {
p.setConfig(config);
p.setText("%pyspark sqlContext.udf.register(\"f1\", lambda x: len(x))\n" +
"sqlContext.sql(\"select f1(\\\"abc\\\") as len\").collect()");
+ p.setAuthenticationInfo(anonymous);
note.run(p.getId());
waitForFinish(p);
assertEquals(Status.FINISHED, p.getStatus());
@@ -242,6 +259,7 @@ public void pySparkTest() throws IOException {
p.setText("%pyspark from pyspark.sql import Row\n" +
"df=sqlContext.createDataFrame([Row(id=1, age=20)])\n" +
"df.collect()");
+ p.setAuthenticationInfo(anonymous);
note.run(p.getId());
waitForFinish(p);
assertEquals(Status.FINISHED, p.getStatus());
@@ -255,19 +273,20 @@ public void pySparkTest() throws IOException {
// use SQLContext to register UDF but use this UDF through SparkSession
p.setText("%pyspark sqlContext.udf.register(\"f1\", lambda x: len(x))\n" +
"spark.sql(\"select f1(\\\"abc\\\") as len\").collect()");
+ p.setAuthenticationInfo(anonymous);
note.run(p.getId());
waitForFinish(p);
assertEquals(Status.FINISHED, p.getStatus());
assertEquals("[Row(len=u'3')]\n", p.getResult().message());
}
}
- ZeppelinServer.notebook.removeNote(note.getId(), null);
+ ZeppelinServer.notebook.removeNote(note.getId(), anonymous);
}
@Test
public void pySparkAutoConvertOptionTest() throws IOException {
// create new note
- Note note = ZeppelinServer.notebook.createNote(null);
+ Note note = ZeppelinServer.notebook.createNote(anonymous);
note.setName("note");
int sparkVersionNumber = getSparkVersionNumber(note);
@@ -286,33 +305,37 @@ public void pySparkAutoConvertOptionTest() throws IOException {
p.setText("%pyspark\nfrom pyspark.sql.functions import *\n"
+ "print(" + sqlContextName + ".range(0, 10).withColumn('uniform', rand(seed=10) * 3.14).count())");
+ p.setAuthenticationInfo(anonymous);
note.run(p.getId());
waitForFinish(p);
assertEquals(Status.FINISHED, p.getStatus());
assertEquals("10\n", p.getResult().message());
}
- ZeppelinServer.notebook.removeNote(note.getId(), null);
+ ZeppelinServer.notebook.removeNote(note.getId(), anonymous);
}
@Test
public void zRunTest() throws IOException {
// create new note
- Note note = ZeppelinServer.notebook.createNote(null);
+ Note note = ZeppelinServer.notebook.createNote(anonymous);
Paragraph p0 = note.addParagraph();
Map config0 = p0.getConfig();
config0.put("enabled", true);
p0.setConfig(config0);
p0.setText("%spark z.run(1)");
+ p0.setAuthenticationInfo(anonymous);
Paragraph p1 = note.addParagraph();
Map config1 = p1.getConfig();
config1.put("enabled", true);
p1.setConfig(config1);
p1.setText("%spark val a=10");
+ p1.setAuthenticationInfo(anonymous);
Paragraph p2 = note.addParagraph();
Map config2 = p2.getConfig();
config2.put("enabled", true);
p2.setConfig(config2);
p2.setText("%spark print(a)");
+ p2.setAuthenticationInfo(anonymous);
note.run(p0.getId());
waitForFinish(p0);
@@ -323,13 +346,13 @@ public void zRunTest() throws IOException {
assertEquals(Status.FINISHED, p2.getStatus());
assertEquals("10", p2.getResult().message());
- ZeppelinServer.notebook.removeNote(note.getId(), null);
+ ZeppelinServer.notebook.removeNote(note.getId(), anonymous);
}
@Test
public void pySparkDepLoaderTest() throws IOException {
// create new note
- Note note = ZeppelinServer.notebook.createNote(null);
+ Note note = ZeppelinServer.notebook.createNote(anonymous);
int sparkVersionNumber = getSparkVersionNumber(note);
if (isPyspark() && sparkVersionNumber >= 14) {
@@ -350,6 +373,7 @@ public void pySparkDepLoaderTest() throws IOException {
config.put("enabled", true);
p0.setConfig(config);
p0.setText("%dep z.load(\"com.databricks:spark-csv_2.11:1.2.0\")");
+ p0.setAuthenticationInfo(anonymous);
note.run(p0.getId());
waitForFinish(p0);
assertEquals(Status.FINISHED, p0.getStatus());
@@ -370,13 +394,14 @@ public void pySparkDepLoaderTest() throws IOException {
"from pyspark.sql import SQLContext\n" +
"print(" + sqlContextName + ".read.format('com.databricks.spark.csv')" +
".load('"+ tmpFile.getAbsolutePath() +"').count())");
+ p1.setAuthenticationInfo(anonymous);
note.run(p1.getId());
waitForFinish(p1);
assertEquals(Status.FINISHED, p1.getStatus());
assertEquals("2\n", p1.getResult().message());
}
- ZeppelinServer.notebook.removeNote(note.getId(), null);
+ ZeppelinServer.notebook.removeNote(note.getId(), anonymous);
}
/**
@@ -390,6 +415,7 @@ private int getSparkVersionNumber(Note note) {
config.put("enabled", true);
p.setConfig(config);
p.setText("%spark print(sc.version)");
+ p.setAuthenticationInfo(anonymous);
note.run(p.getId());
waitForFinish(p);
assertEquals(Status.FINISHED, p.getStatus());
diff --git a/zeppelin-server/src/test/java/org/apache/zeppelin/socket/NotebookServerTest.java b/zeppelin-server/src/test/java/org/apache/zeppelin/socket/NotebookServerTest.java
index 01a24e2e1de..caac04349e2 100644
--- a/zeppelin-server/src/test/java/org/apache/zeppelin/socket/NotebookServerTest.java
+++ b/zeppelin-server/src/test/java/org/apache/zeppelin/socket/NotebookServerTest.java
@@ -31,6 +31,7 @@
import org.apache.zeppelin.notebook.socket.Message.OP;
import org.apache.zeppelin.rest.AbstractTestRestApi;
import org.apache.zeppelin.server.ZeppelinServer;
+import org.apache.zeppelin.user.AuthenticationInfo;
import org.junit.AfterClass;
import org.junit.Before;
import org.junit.BeforeClass;
@@ -57,6 +58,7 @@ public class NotebookServerTest extends AbstractTestRestApi {
private static NotebookServer notebookServer;
private static Gson gson;
private HttpServletRequest mockRequest;
+ private AuthenticationInfo anonymous;
@BeforeClass
public static void init() throws Exception {
@@ -74,6 +76,7 @@ public static void destroy() throws Exception {
@Before
public void setUp() {
mockRequest = mock(HttpServletRequest.class);
+ anonymous = new AuthenticationInfo("anonymous");
}
@Test
@@ -94,14 +97,14 @@ public void checkInvalidOrigin(){
@Test
public void testMakeSureNoAngularObjectBroadcastToWebsocketWhoFireTheEvent() throws IOException {
// create a notebook
- Note note1 = notebook.createNote(null);
+ Note note1 = notebook.createNote(anonymous);
// get reference to interpreterGroup
InterpreterGroup interpreterGroup = null;
List settings = notebook.getInterpreterFactory().getInterpreterSettings(note1.getId());
for (InterpreterSetting setting : settings) {
if (setting.getName().equals("md")) {
- interpreterGroup = setting.getInterpreterGroup("sharedProcess");
+ interpreterGroup = setting.getInterpreterGroup("anonymous", "sharedProcess");
break;
}
}
@@ -109,6 +112,7 @@ public void testMakeSureNoAngularObjectBroadcastToWebsocketWhoFireTheEvent() thr
// start interpreter process
Paragraph p1 = note1.addParagraph();
p1.setText("%md start remote interpreter process");
+ p1.setAuthenticationInfo(anonymous);
note1.run(p1.getId());
// add angularObject
@@ -144,7 +148,7 @@ public void testMakeSureNoAngularObjectBroadcastToWebsocketWhoFireTheEvent() thr
verify(sock1, times(0)).send(anyString());
verify(sock2, times(1)).send(anyString());
- notebook.removeNote(note1.getId(), null);
+ notebook.removeNote(note1.getId(), anonymous);
}
@Test
@@ -167,7 +171,7 @@ public void testImportNotebook() throws IOException {
assertNotEquals(null, notebook.getNote(note.getId()));
assertEquals("Test Zeppelin notebook import", notebook.getNote(note.getId()).getName());
assertEquals("Test paragraphs import", notebook.getNote(note.getId()).getParagraphs().get(0).getText());
- notebook.removeNote(note.getId(), null);
+ notebook.removeNote(note.getId(), anonymous);
}
@Test
diff --git a/zeppelin-web/bower.json b/zeppelin-web/bower.json
index aeabd79dca2..6a101b8e1f2 100644
--- a/zeppelin-web/bower.json
+++ b/zeppelin-web/bower.json
@@ -33,7 +33,6 @@
"handsontable": "~0.24.2",
"moment-duration-format": "^1.3.0",
"select2": "^4.0.3",
- "angular-esri-map": "~2.0.0",
"github-markdown-css": "^2.4.0"
},
"devDependencies": {
diff --git a/zeppelin-web/src/app/app.js b/zeppelin-web/src/app/app.js
index 603171f10a6..30d24c2d2eb 100644
--- a/zeppelin-web/src/app/app.js
+++ b/zeppelin-web/src/app/app.js
@@ -33,8 +33,7 @@
'xeditable',
'ngToast',
'focus-if',
- 'ngResource',
- 'esri.map'
+ 'ngResource'
])
.filter('breakFilter', function() {
return function(text) {
diff --git a/zeppelin-web/src/app/configuration/configuration.html b/zeppelin-web/src/app/configuration/configuration.html
index 0bee84d7939..c0ec367b3e6 100644
--- a/zeppelin-web/src/app/configuration/configuration.html
+++ b/zeppelin-web/src/app/configuration/configuration.html
@@ -11,6 +11,7 @@
See the License for the specific language governing permissions and
limitations under the License.
-->
+
-
+
Keys
@@ -165,52 +165,4 @@
-
-
diff --git a/zeppelin-web/src/app/notebook/paragraph/paragraph.controller.js b/zeppelin-web/src/app/notebook/paragraph/paragraph.controller.js
index 6eaba8282db..deab258ed65 100644
--- a/zeppelin-web/src/app/notebook/paragraph/paragraph.controller.js
+++ b/zeppelin-web/src/app/notebook/paragraph/paragraph.controller.js
@@ -30,13 +30,12 @@
'websocketMsgSrv',
'baseUrlSrv',
'ngToast',
- 'saveAsService',
- 'esriLoader'
+ 'saveAsService'
];
function ParagraphCtrl($scope, $rootScope, $route, $window, $routeParams, $location,
$timeout, $compile, $http, $q, websocketMsgSrv,
- baseUrlSrv, ngToast, saveAsService, esriLoader) {
+ baseUrlSrv, ngToast, saveAsService) {
var ANGULAR_FUNCTION_OBJECT_NAME_PREFIX = '_Z_ANGULAR_FUNC_';
$scope.parentNote = null;
$scope.paragraph = null;
@@ -105,7 +104,6 @@
$scope.parentNote = note;
$scope.originalText = angular.copy(newParagraph.text);
$scope.chart = {};
- $scope.baseMapOption = ['Streets', 'Satellite', 'Hybrid', 'Topo', 'Gray', 'Oceans', 'Terrain'];
$scope.colWidthOption = [1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12];
$scope.paragraphFocused = false;
if (newParagraph.focus) {
@@ -257,22 +255,6 @@
config.graph.scatter = {};
}
- if (!config.graph.map) {
- config.graph.map = {};
- }
-
- if (!config.graph.map.baseMapType) {
- config.graph.map.baseMapType = $scope.baseMapOption[0];
- }
-
- if (!config.graph.map.isOnline) {
- config.graph.map.isOnline = true;
- }
-
- if (!config.graph.map.pinCols) {
- config.graph.map.pinCols = [];
- }
-
if (config.enabled === undefined) {
config.enabled = true;
}
@@ -974,8 +956,6 @@
if (!type || type === 'table') {
setTable($scope.paragraph.result, refresh);
- } else if (type === 'map') {
- setMap($scope.paragraph.result, refresh);
} else {
setD3Chart(type, $scope.paragraph.result, refresh);
}
@@ -1206,236 +1186,6 @@
$timeout(retryRenderer);
};
- var setMap = function(data, refresh) {
- var createPinMapLayer = function(pins, cb) {
- esriLoader.require(['esri/layers/FeatureLayer'], function(FeatureLayer) {
- var pinLayer = new FeatureLayer({
- id: 'pins',
- spatialReference: $scope.map.spatialReference,
- geometryType: 'point',
- source: pins,
- fields: [],
- objectIdField: '_ObjectID',
- renderer: $scope.map.pinRenderer,
- popupTemplate: {
- title: '[{_lng}, {_lat}]',
- content: [{
- type: 'fields',
- fieldInfos: []
- }]
- }
- });
-
- // add user-selected pin info fields to popup
- var pinInfoCols = $scope.paragraph.config.graph.map.pinCols;
- for (var i = 0; i < pinInfoCols.length; ++i) {
- pinLayer.popupTemplate.content[0].fieldInfos.push({
- fieldName: pinInfoCols[i].name,
- visible: true
- });
- }
- cb(pinLayer);
- });
- };
-
- var getMapPins = function(cb) {
- esriLoader.require(['esri/geometry/Point'], function(Point, FeatureLayer) {
- var latCol = $scope.paragraph.config.graph.map.lat;
- var lngCol = $scope.paragraph.config.graph.map.lng;
- var pinInfoCols = $scope.paragraph.config.graph.map.pinCols;
- var pins = [];
-
- // construct objects for pins
- if (latCol && lngCol && data.rows) {
- for (var i = 0; i < data.rows.length; ++i) {
- var row = data.rows[i];
- var lng = row[lngCol.index];
- var lat = row[latCol.index];
- var pin = {
- geometry: new Point({
- longitude: lng,
- latitude: lat,
- spatialReference: $scope.map.spatialReference
- }),
- attributes: {
- _ObjectID: i,
- _lng: lng,
- _lat: lat
- }
- };
-
- // add pin info from user-selected columns
- for (var j = 0; j < pinInfoCols.length; ++j) {
- var col = pinInfoCols[j];
- pin.attributes[col.name] = row[col.index];
- }
- pins.push(pin);
- }
- }
- cb(pins);
- });
- };
-
- var updateMapPins = function() {
- var pinLayer = $scope.map.map.findLayerById('pins');
- $scope.map.popup.close();
- if (pinLayer) {
- $scope.map.map.remove(pinLayer);
- }
-
- // add pins to map as layer
- getMapPins(function(pins) {
- createPinMapLayer(pins, function(pinLayer) {
- $scope.map.map.add(pinLayer);
- if (pinLayer.source.length > 0) {
- $scope.map.goTo(pinLayer.source);
- }
- });
- });
- };
-
- var createMap = function(mapdiv) {
- // prevent zooming with the scroll wheel
- var disableZoom = function(e) {
- var evt = e || window.event;
- evt.cancelBubble = true;
- evt.returnValue = false;
- if (evt.stopPropagation) {
- evt.stopPropagation();
- }
- };
- var eName = window.WheelEvent ? 'wheel' : // Modern browsers
- window.MouseWheelEvent ? 'mousewheel' : // WebKit and IE
- 'DOMMouseScroll'; // Old Firefox
- mapdiv.addEventListener(eName, disableZoom, true);
-
- esriLoader.require(['esri/views/MapView',
- 'esri/Map',
- 'esri/renderers/SimpleRenderer',
- 'esri/symbols/SimpleMarkerSymbol'],
- function(MapView, Map, SimpleRenderer, SimpleMarkerSymbol) {
- $scope.map = new MapView({
- container: mapdiv,
- map: new Map({
- basemap: $scope.paragraph.config.graph.map.baseMapType.toLowerCase()
- }),
- center: [-106.3468, 56.1304], // Canada (lng, lat)
- zoom: 2,
- pinRenderer: new SimpleRenderer({
- symbol: new SimpleMarkerSymbol({
- 'color': [255, 0, 0, 0.5],
- 'size': 16.5,
- 'outline': {
- 'color': [0, 0, 0, 1],
- 'width': 1.125,
- },
- // map pin SVG path
- 'path': 'M16,3.5c-4.142,0-7.5,3.358-7.5,7.5c0,4.143,7.5,18.121,7.5,' +
- '18.121S23.5,15.143,23.5,11C23.5,6.858,20.143,3.5,16,3.5z ' +
- 'M16,14.584c-1.979,0-3.584-1.604-3.584-3.584S14.021,7.416,' +
- '16,7.416S19.584,9.021,19.584,11S17.979,14.584,16,14.584z'
- })
- })
- });
-
- $scope.map.on('click', function() {
- // ArcGIS JS API 4.0 does not account for scrolling or position
- // changes by default (this is a bug, to be fixed in the upcoming
- // version 4.1; see https://geonet.esri.com/thread/177238#comment-609681).
- // This results in a misaligned popup.
-
- // Workaround: manually set popup position to match position of selected pin
- if ($scope.map.popup.selectedFeature) {
- $scope.map.popup.location = $scope.map.popup.selectedFeature.geometry;
- }
- });
- $scope.map.then(updateMapPins);
- });
- };
-
- var checkMapOnline = function(cb) {
- // are we able to get a response from the ArcGIS servers?
- var callback = function(res) {
- var online = (res.status > 0);
- $scope.paragraph.config.graph.map.isOnline = online;
- cb(online);
- };
- $http.head('//services.arcgisonline.com/arcgis/', {
- timeout: 5000,
- withCredentials: false
- }).then(callback, callback);
- };
-
- var renderMap = function() {
- var mapdiv = angular.element('#p' + $scope.paragraph.id + '_map')
- .css('height', $scope.paragraph.config.graph.height)
- .children('div').get(0);
-
- // on chart type change, destroy map to force reinitialization.
- if ($scope.map && !refresh) {
- $scope.map.map.destroy();
- $scope.map.pinRenderer = null;
- $scope.map = null;
- }
-
- var requireMapCSS = function() {
- var url = '//js.arcgis.com/4.0/esri/css/main.css';
- if (!angular.element('link[href="' + url + '"]').length) {
- var link = document.createElement('link');
- link.rel = 'stylesheet';
- link.type = 'text/css';
- link.href = url;
- angular.element('head').append(link);
- }
- };
-
- var requireMapJS = function(cb) {
- if (!esriLoader.isLoaded()) {
- esriLoader.bootstrap({
- url: '//js.arcgis.com/4.0'
- }).then(cb);
- } else {
- cb();
- }
- };
-
- checkMapOnline(function(online) {
- // we need an internet connection to use the map
- if (online) {
- // create map if not exists.
- if (!$scope.map) {
- requireMapCSS();
- requireMapJS(function() {
- createMap(mapdiv);
- });
- } else {
- updateMapPins();
- }
- }
- });
- };
-
- var retryRenderer = function() {
- if (angular.element('#p' + $scope.paragraph.id + '_map div').length) {
- try {
- renderMap();
- } catch (err) {
- console.log('Map drawing error %o', err);
- }
- } else {
- $timeout(retryRenderer,10);
- }
- };
- $timeout(retryRenderer);
- };
-
- $scope.setMapBaseMap = function(bm) {
- $scope.paragraph.config.graph.map.baseMapType = bm;
- if ($scope.map) {
- $scope.map.map.basemap = bm.toLowerCase();
- }
- };
-
$scope.isGraphMode = function(graphName) {
var activeAppId = _.get($scope.paragraph.config, 'helium.activeApp');
if ($scope.getResultType() === 'TABLE' && $scope.getGraphMode() === graphName && !activeAppId) {
@@ -1498,24 +1248,6 @@
$scope.setGraphMode($scope.paragraph.config.graph.mode, true, false);
};
- $scope.removeMapOptionLat = function(idx) {
- $scope.paragraph.config.graph.map.lat = null;
- clearUnknownColsFromGraphOption();
- $scope.setGraphMode($scope.paragraph.config.graph.mode, true, false);
- };
-
- $scope.removeMapOptionLng = function(idx) {
- $scope.paragraph.config.graph.map.lng = null;
- clearUnknownColsFromGraphOption();
- $scope.setGraphMode($scope.paragraph.config.graph.mode, true, false);
- };
-
- $scope.removeMapOptionPinInfo = function(idx) {
- $scope.paragraph.config.graph.map.pinCols.splice(idx, 1);
- clearUnknownColsFromGraphOption();
- $scope.setGraphMode($scope.paragraph.config.graph.mode, true, false);
- };
-
/* Clear unknown columns from graph option */
var clearUnknownColsFromGraphOption = function() {
var unique = function(list) {
@@ -1574,10 +1306,6 @@
removeUnknown($scope.paragraph.config.graph.groups);
removeUnknownFromFields($scope.paragraph.config.graph.scatter);
-
- unique($scope.paragraph.config.graph.map.pinCols);
- removeUnknown($scope.paragraph.config.graph.map.pinCols);
- removeUnknownFromFields($scope.paragraph.config.graph.map);
};
/* select default key and value if there're none selected */
@@ -1598,23 +1326,6 @@
$scope.paragraph.config.graph.scatter.xAxis = $scope.paragraph.result.columnNames[0];
}
}
-
- /* try to find columns for the map logitude and latitude */
- var findDefaultMapCol = function(settingName, keyword) {
- var col;
- if (!$scope.paragraph.config.graph.map[settingName]) {
- for (var i = 0; i < $scope.paragraph.result.columnNames.length; ++i) {
- col = $scope.paragraph.result.columnNames[i];
- if (col.name.toUpperCase().indexOf(keyword) !== -1) {
- $scope.paragraph.config.graph.map[settingName] = col;
- break;
- }
- }
- }
- };
-
- findDefaultMapCol('lat', 'LAT');
- findDefaultMapCol('lng', 'LONG');
};
var pivot = function(data) {
diff --git a/zeppelin-web/src/app/notebook/paragraph/paragraph.css b/zeppelin-web/src/app/notebook/paragraph/paragraph.css
index 0b821c94181..60b93d0f9a1 100644
--- a/zeppelin-web/src/app/notebook/paragraph/paragraph.css
+++ b/zeppelin-web/src/app/notebook/paragraph/paragraph.css
@@ -331,41 +331,12 @@ table.dataTable.table-condensed .sorting_desc:after {
.tableDisplay div {
}
-.tableDisplay img:not(.esri-bitmap) {
+.tableDisplay img {
display: block;
max-width: 100%;
height: auto;
}
-.esri-display-object > svg {
- overflow: visible;
-}
-
-.esri-popup > .esri-docked.esri-dock-to-bottom {
- padding: 8px;
- margin-top: 0px;
-}
-
-.esri-popup-main {
- max-height: 100%;
-}
-
-span.map-offline-text {
- display: table;
- width: 100%;
- height: 100%;
- text-align: center;
-}
-
-span.map-offline-text > span {
- display: table-cell;
- vertical-align: middle;
- font-size: 18px;
- font-weight: 700;
- font-family: "Helvetica Neue", Helvetica, Arial, sans-serif;
- color: #212121;
-}
-
.tableDisplay .btn-group span {
margin: 10px 0 0 10px;
font-size: 12px;
@@ -383,8 +354,7 @@ span.map-offline-text > span {
}
-.tableDisplay .option .columns,
-div.esri-view {
+.tableDisplay .option .columns {
height: 100%;
}
diff --git a/zeppelin-web/src/components/navbar/navbar.controller.js b/zeppelin-web/src/components/navbar/navbar.controller.js
index 7838a824e20..8586960af1c 100644
--- a/zeppelin-web/src/components/navbar/navbar.controller.js
+++ b/zeppelin-web/src/components/navbar/navbar.controller.js
@@ -57,6 +57,7 @@
}
function initController() {
+ $scope.isDrawNavbarNotebookList = false;
angular.element('#notebook-list').perfectScrollbar({suppressScrollX: true});
angular.element(document).click(function() {
@@ -122,6 +123,19 @@
$scope.$on('loginSuccess', function(event, param) {
loadNotes();
});
+
+ /*
+ ** Performance optimization for Browser Render.
+ */
+ angular.element(document).ready(function() {
+ angular.element('.notebook-list-dropdown').on('show.bs.dropdown', function() {
+ $scope.isDrawNavbarNotebookList = true;
+ });
+
+ angular.element('.notebook-list-dropdown').on('hide.bs.dropdown', function() {
+ $scope.isDrawNavbarNotebookList = false;
+ });
+ });
}
})();
diff --git a/zeppelin-web/src/components/navbar/navbar.html b/zeppelin-web/src/components/navbar/navbar.html
index c4de8468076..8851409b025 100644
--- a/zeppelin-web/src/components/navbar/navbar.html
+++ b/zeppelin-web/src/components/navbar/navbar.html
@@ -25,12 +25,12 @@
+
+
+## (Experimental) Interpreter Execution Hooks
+
+Zeppelin allows for users to specify additional code to be executed by an interpreter at pre and post-paragraph code execution. This is primarily useful if you need to run the same set of code for all of the paragraphs within your notebook at specific times. Currently, this feature is only available for the spark and pyspark interpreters. To specify your hook code, you may use '`z.registerHook()`. For example, enter the following into one paragraph:
+
+```python
+%pyspark
+z.registerHook("post_exec", "print 'This code should be executed before the parapgraph code!'")
+z.registerHook("pre_exec", "print 'This code should be executed after the paragraph code!'")
+```
+
+These calls will not take into effect until the next time you run a paragraph. In another paragraph, enter
+```python
+%pyspark
+print "This code should be entered into the paragraph by the user!"
+```
+
+The output should be:
+```
+This code should be executed before the paragraph code!
+This code should be entered into the paragraph by the user!
+This code should be executed after the paragraph code!
+```
+
+If you ever need to know the hook code, use `z.getHook()`:
+```python
+%pyspark
+print z.getHook("post_exec")
+```
+```
+print 'This code should be executed after the paragraph code!'
+```
+Any call to `z.registerHook()` will automatically overwrite what was previously registered. To completely unregister a hook event, use `z.unregisterHook(eventCode)`. Currently only `"post_exec"` and `"pre_exec"` are valid event codes for the Zeppelin Hook Registry system.
+
+Finally, the hook registry is internally shared by other interpreters in the same group. This would allow for hook code for one interpreter REPL to be set by another as follows:
+
+```scala
+%spark
+z.unregisterHook("post_exec", "pyspark")
+```
+The API is identical for both the spark (scala) and pyspark (python) implementations.
+
+### Caveats
+Calls to `z.registerHook("pre_exec", ...)` should be made with care. If there are errors in your specified hook code, this will cause the interpreter REPL to become unable to execute any code pass the pre-execute stage making it impossible for direct calls to `z.unregisterHook()` to take into effect. Current workarounds include calling `z.unregisterHook()` from a different interpreter REPL in the same interpreter group (see above) or manually restarting the interpreter group in the UI.
+
diff --git a/elasticsearch/src/main/java/org/apache/zeppelin/elasticsearch/ElasticsearchInterpreter.java b/elasticsearch/src/main/java/org/apache/zeppelin/elasticsearch/ElasticsearchInterpreter.java
index 511f1b5e38b..549b5f25efd 100644
--- a/elasticsearch/src/main/java/org/apache/zeppelin/elasticsearch/ElasticsearchInterpreter.java
+++ b/elasticsearch/src/main/java/org/apache/zeppelin/elasticsearch/ElasticsearchInterpreter.java
@@ -22,6 +22,7 @@
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
+import java.util.HashSet;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
@@ -35,7 +36,6 @@
import org.apache.commons.lang.StringUtils;
import org.apache.zeppelin.interpreter.Interpreter;
import org.apache.zeppelin.interpreter.InterpreterContext;
-import org.apache.zeppelin.interpreter.InterpreterPropertyBuilder;
import org.apache.zeppelin.interpreter.InterpreterResult;
import org.apache.zeppelin.interpreter.thrift.InterpreterCompletion;
import org.elasticsearch.action.delete.DeleteResponse;
@@ -48,6 +48,8 @@
import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.transport.InetSocketTransportAddress;
+import org.elasticsearch.common.xcontent.XContentBuilder;
+import org.elasticsearch.common.xcontent.XContentFactory;
import org.elasticsearch.common.xcontent.XContentHelper;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.search.SearchHit;
@@ -437,14 +439,37 @@ else if (agg instanceof InternalSingleBucketAggregation) {
resMsg = XContentHelper.toString((InternalSingleBucketAggregation) agg).toString();
}
else if (agg instanceof InternalMultiBucketAggregation) {
- final StringBuffer buffer = new StringBuffer("key\tdoc_count");
-
+ final Set