Elgato_dark/zeppelin
1
0
Fork 0
mirror of https://github.com/apache/zeppelin synced 2026-05-24 09:38:26 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 1

Merge branch 'ZEPPELIN-3585' of https://github.com/TinkoffCreditSystems/zeppelin into ZEPPELIN-3585

Browse source 
# Conflicts:
#	zeppelin-server/src/main/java/org/apache/zeppelin/rest/NotebookRestApi.java
This commit is contained in:
oxygen311 2018-07-19 12:23:16 +03:00
commit 62d081c9d3
76 changed files with 1361 additions and 465 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
.travis.yml
View file

@ -17,6 +17,10 @@ language: java
sudo: false
before_cache:
- sudo chown -R travis:travis $HOME/.m2
cache:
apt: true
directories:

6
bin/interpreter.sh
View file

@ -92,7 +92,7 @@ HOSTNAME=$(hostname)
ZEPPELIN_SERVER=org.apache.zeppelin.interpreter.remote.RemoteInterpreterServer
INTERPRETER_ID=$(basename "${INTERPRETER_DIR}")
ZEPPELIN_PID="${ZEPPELIN_PID_DIR}/zeppelin-interpreter-${INTERPRETER_ID}-${ZEPPELIN_IDENT_STRING}-${HOSTNAME}.pid"
ZEPPELIN_PID="${ZEPPELIN_PID_DIR}/zeppelin-interpreter-${INTERPRETER_ID}-${ZEPPELIN_IDENT_STRING}-${HOSTNAME}-${PORT}.pid"
ZEPPELIN_LOGFILE="${ZEPPELIN_LOG_DIR}/zeppelin-interpreter-${INTERPRETER_SETTING_NAME}-"
if [[ -z "$ZEPPELIN_IMPERSONATE_CMD" ]]; then
@ -251,14 +251,14 @@ function shutdown_hook() {
sleep 3
let "count+=1"
else
rm -f "${ZEPPELIN_PID}"
break
fi
if [[ "${count}" == "5" ]]; then
$(kill -9 ${pid} > /dev/null 2> /dev/null)
rm -f "${ZEPPELIN_PID}"
fi
done
}
wait
rm -f "${ZEPPELIN_PID}" > /dev/null 2> /dev/null

4
cassandra/src/main/scala/org/apache/zeppelin/cassandra/ParagraphParser.scala
View file

@ -118,12 +118,12 @@ object ParagraphParser {
val DESCRIBE_AGGREGATES_PATTERN = ("""^(?i)\s*(?:DESCRIBE|DESC)\s+AGGREGATES\s*;\s*$""").r
val DESCRIBE_MATERIALIZED_VIEW_PATTERN = ("""^(?i)\s*(?:DESCRIBE|DESC)\s+MATERIALIZED\s+VIEW\s*("""+VALID_IDENTIFIER+""");\s*$""").r
val DESCRIBE_MATERIALIZED_VIEW_PATTERN = ("""^(?i)\s*(?:DESCRIBE|DESC)\s+MATERIALIZED\s+VIEW\s*("""+VALID_IDENTIFIER+""")\s*;\s*$""").r
val DESCRIBE_MATERIALIZED_VIEW_WITH_KEYSPACE_PATTERN = ("""^(?i)\s*(?:DESCRIBE|DESC)\s+MATERIALIZED\s+VIEW\s*(""" +
VALID_IDENTIFIER +
""")\.(""" +
VALID_IDENTIFIER +
""");\s*$""").r
""")\s*;\s*$""").r
val DESCRIBE_MATERIALIZED_VIEWS_PATTERN = ("""^(?i)\s*(?:DESCRIBE|DESC)\s+MATERIALIZED\s+VIEWS\s*;\s*$""").r

1
conf/log4j.properties
View file

@ -22,7 +22,6 @@ log4j.appender.stdout.layout = org.apache.log4j.PatternLayout
log4j.appender.stdout.layout.ConversionPattern=%5p [%d] ({%t} %F[%M]:%L) - %m%n
log4j.appender.dailyfile.DatePattern=.yyyy-MM-dd
log4j.appender.dailyfile.Threshold = INFO
log4j.appender.dailyfile = org.apache.log4j.DailyRollingFileAppender
log4j.appender.dailyfile.File = ${zeppelin.log.file}
log4j.appender.dailyfile.layout = org.apache.log4j.PatternLayout

1
conf/shiro.ini.template
View file

@ -116,5 +116,6 @@ admin = *
/api/interpreter/** = authc, roles[admin]
/api/configurations/** = authc, roles[admin]
/api/credential/** = authc, roles[admin]
/api/admin/** = authc, roles[admin]
#/** = anon
/** = authc

20
docs/interpreter/spark.md
View file

@ -120,6 +120,11 @@ You can also set other Spark properties which are not listed in the table. For a
<td>false</td>
<td>Execute multiple SQL concurrently if set true.</td>
</tr>
<tr>
<td>zeppelin.spark.concurrentSQL.max</td>
<td>10</td>
<td>Max number of SQL concurrently executed</td>
</tr>
<tr>
<td>zeppelin.spark.maxResult</td>
<td>1000</td>
@ -332,6 +337,21 @@ utilizing Zeppelin's built-in [Angular Display System](../usage/display_system/a
<img class="img-responsive" src="{{BASE_PATH}}/assets/themes/zeppelin/img/docs-img/matplotlibAngularExample.gif" />
## Running spark sql concurrently
By default, each sql statement would run sequentially in `%spark.sql`. But you can run them concurrently by following setup.
1. set `zeppelin.spark.concurrentSQL` to true to enable the sql concurrent feature, underneath zeppelin will change to use fairscheduler for spark. And also set `zeppelin.spark.concurrentSQL.max` to control the max number of sql statements running concurrently.
2. configure pools by creating `fairscheduler.xml` under your `SPARK_CONF_DIR`, check the offical spark doc [Configuring Pool Properties](http://spark.apache.org/docs/latest/job-scheduling.html#configuring-pool-properties)
3. set pool property via setting paragraph property. e.g.
```
%spark(pool=pool1)
sql statement
```
This feature is available for both all versions of scala spark, pyspark. For sparkr, it is only available starting from 2.3.0.
## Interpreter setting option
You can choose one of `shared`, `scoped` and `isolated` options wheh you configure Spark interpreter.

2
docs/quickstart/install.md
View file

@ -36,7 +36,7 @@ Apache Zeppelin officially supports and is tested on the following environments:
</tr>
<tr>
<td>Oracle JDK</td>
<td>1.7 <br /> (set <code>JAVA_HOME</code>)</td>
<td>1.8 (171) <br /> (set <code>JAVA_HOME</code>)</td>
</tr>
<tr>
<td>OS</td>

4
docs/setup/basics/multi_user_support.md
View file

@ -25,8 +25,8 @@ limitations under the License.
This page describes about multi-user support.
- multiple users login / logout using [Shiro Authentication](../setup/security/shiro_authentication.html)
- managing [Notebook Permission](../setup/security/notebook_authorization.html)
- multiple users login / logout using [Shiro Authentication](../security/shiro_authentication.html)
- managing [Notebook Permission](../security/notebook_authorization.html)
- how to setup [impersonation for interpreters](../../usage/interpreter/user_impersonation.html)
- different contexts per user / note using [Interpreter Binding Mode](../../usage/interpreter/interpreter_binding_mode.html)
- a paragraph in a notebook can be [Personalized](../../usage/other_features/personalized_mode.html)

4
docs/setup/deployment/yarn_install.md
View file

@ -45,7 +45,7 @@ Its assumed in the rest of the document that zeppelin user is indeed created and
### List of Prerequisites
* CentOS 6.x, Mac OSX, Ubuntu 14.X
* Java 1.7
* Java 1.8
* Hadoop client
* Spark
* Internet connection is required.
@ -89,7 +89,7 @@ cp /home/zeppelin/zeppelin/conf/zeppelin-env.sh.template /home/zeppelin/zeppelin
Set the following properties
```bash
export JAVA_HOME="/usr/java/jdk1.7.0_79"
export JAVA_HOME="/usr/java/jdk1.8.0_171"
export HADOOP_CONF_DIR="/etc/hadoop/conf"
export ZEPPELIN_JAVA_OPTS="-Dhdp.version=2.3.1.0-2574"
export SPARK_HOME="/usr/lib/spark"

4
docs/setup/storage/storage.md
View file

@ -204,6 +204,7 @@ Or using the following setting in **zeppelin-site.xml**:
```
</br>
## Notebook Storage in Azure <a name="Azure"></a>
Using `AzureNotebookRepo` you can connect your Zeppelin with your Azure account for notebook storage.
@ -265,7 +266,8 @@ Optionally, you can specify Azure folder structure name in the file **zeppelin-s
```
</br>
## Notebook Storage in Google Cloud Storage<a name="GCS"></a>
## Notebook Storage in Google Cloud Storage <a name="GCS"></a>
Using `GCSNotebookRepo` you can connect Zeppelin with Google Cloud Storage using [Application Default Credentials](https://cloud.google.com/docs/authentication/production).

3
jdbc/src/main/resources/interpreter-setting.json
View file

@ -57,7 +57,8 @@
"envName": null,
"propertyName": "default.statementPrecode",
"defaultValue": "",
"description": "Runs before each run of the paragraph, in the same connection"
"description": "Runs before each run of the paragraph, in the same connection",
"type": "textarea"
},
"default.splitQueries": {
"envName": null,

4
python/src/main/java/org/apache/zeppelin/python/IPythonInterpreter.java
View file

@ -178,7 +178,7 @@ public class IPythonInterpreter extends Interpreter implements ExecuteResultHand
return "jupyter-client is not installed.";
}
if (!freezeOutput.contains("ipykernel=")) {
return "ipkernel is not installed";
return "ipykernel is not installed";
}
if (!freezeOutput.contains("ipython=")) {
return "ipython is not installed";
@ -186,7 +186,7 @@ public class IPythonInterpreter extends Interpreter implements ExecuteResultHand
if (!freezeOutput.contains("grpcio=")) {
return "grpcio is not installed";
}
LOGGER.info("IPython prerequisite is meet");
LOGGER.info("IPython prerequisite is met");
} catch (Exception e) {
LOGGER.warn("Fail to checkIPythonPrerequisite", e);
return "Fail to checkIPythonPrerequisite: " + ExceptionUtils.getStackTrace(e);

11
spark/interpreter/src/main/java/org/apache/zeppelin/spark/IPySparkInterpreter.java
View file

@ -118,12 +118,21 @@ public class IPySparkInterpreter extends IPythonInterpreter {
public InterpreterResult interpret(String st, InterpreterContext context) {
InterpreterContext.set(context);
String jobGroupId = Utils.buildJobGroupId(context);
String jobDesc = "Started by: " + Utils.getUserName(context.getAuthenticationInfo());
String jobDesc = Utils.buildJobDesc(context);
String setJobGroupStmt = "sc.setJobGroup('" + jobGroupId + "', '" + jobDesc + "')";
InterpreterResult result = super.interpret(setJobGroupStmt, context);
if (result.code().equals(InterpreterResult.Code.ERROR)) {
return new InterpreterResult(InterpreterResult.Code.ERROR, "Fail to setJobGroup");
}
String pool = "None";
if (context.getLocalProperties().containsKey("pool")) {
pool = "'" + context.getLocalProperties().get("pool") + "'";
}
String setPoolStmt = "sc.setLocalProperty('spark.scheduler.pool', " + pool + ")";
result = super.interpret(setPoolStmt, context);
if (result.code().equals(InterpreterResult.Code.ERROR)) {
return new InterpreterResult(InterpreterResult.Code.ERROR, "Fail to setPool");
}
return super.interpret(st, context);
}

13
spark/interpreter/src/main/java/org/apache/zeppelin/spark/NewSparkInterpreter.java
View file

@ -86,9 +86,15 @@ public class NewSparkInterpreter extends AbstractSparkInterpreter {
if (!StringUtils.isBlank(entry.getValue().toString())) {
conf.set(entry.getKey().toString(), entry.getValue().toString());
}
// zeppelin.spark.useHiveContext & zeppelin.spark.concurrentSQL are legacy zeppelin
// properties, convert them to spark properties here.
if (entry.getKey().toString().equals("zeppelin.spark.useHiveContext")) {
conf.set("spark.useHiveContext", entry.getValue().toString());
}
if (entry.getKey().toString().equals("zeppelin.spark.concurrentSQL")
&& entry.getValue().toString().equals("true")) {
conf.set("spark.scheduler.mode", "FAIR");
}
}
// use local mode for embedded spark mode when spark.master is not found
conf.setIfMissing("spark.master", "local");
@ -141,8 +147,11 @@ public class NewSparkInterpreter extends AbstractSparkInterpreter {
z.setGui(context.getGui());
z.setNoteGui(context.getNoteGui());
z.setInterpreterContext(context);
String jobDesc = "Started by: " + Utils.getUserName(context.getAuthenticationInfo());
sc.setJobGroup(Utils.buildJobGroupId(context), jobDesc, false);
sc.setJobGroup(Utils.buildJobGroupId(context), Utils.buildJobDesc(context), false);
// set spark.scheduler.pool to null to clear the pool assosiated with this paragraph
// sc.setLocalProperty("spark.scheduler.pool", null) will clean the pool
sc.setLocalProperty("spark.scheduler.pool", context.getLocalProperties().get("pool"));
return innerInterpreter.interpret(st, context);
}

3
spark/interpreter/src/main/java/org/apache/zeppelin/spark/OldSparkInterpreter.java
View file

@ -1041,8 +1041,7 @@ public class OldSparkInterpreter extends AbstractSparkInterpreter {
synchronized (this) {
z.setGui(context.getGui());
z.setNoteGui(context.getNoteGui());
String jobDesc = "Started by: " + Utils.getUserName(context.getAuthenticationInfo());
sc.setJobGroup(Utils.buildJobGroupId(context), jobDesc, false);
sc.setJobGroup(Utils.buildJobGroupId(context), Utils.buildJobDesc(context), false);
InterpreterResult r = interpretInput(lines, context);
sc.clearJobGroup();
return r;

9
spark/interpreter/src/main/java/org/apache/zeppelin/spark/PySparkInterpreter.java
View file

@ -150,10 +150,17 @@ public class PySparkInterpreter extends PythonInterpreter {
@Override
protected void preCallPython(InterpreterContext context) {
String jobGroup = Utils.buildJobGroupId(context);
String jobDesc = "Started by: " + Utils.getUserName(context.getAuthenticationInfo());
String jobDesc = Utils.buildJobDesc(context);
callPython(new PythonInterpretRequest(
String.format("if 'sc' in locals():\n\tsc.setJobGroup('%s', '%s')", jobGroup, jobDesc),
false, false));
String pool = "None";
if (context.getLocalProperties().containsKey("pool")) {
pool = "'" + context.getLocalProperties().get("pool") + "'";
}
String setPoolStmt = "sc.setLocalProperty('spark.scheduler.pool', " + pool + ")";
callPython(new PythonInterpretRequest(setPoolStmt, false, false));
}
// Run python shell

13
spark/interpreter/src/main/java/org/apache/zeppelin/spark/SparkRInterpreter.java
View file

@ -123,8 +123,7 @@ public class SparkRInterpreter extends Interpreter {
throws InterpreterException {
String jobGroup = Utils.buildJobGroupId(interpreterContext);
String jobDesc = "Started by: " +
Utils.getUserName(interpreterContext.getAuthenticationInfo());
String jobDesc = Utils.buildJobDesc(interpreterContext);
sparkInterpreter.getSparkContext().setJobGroup(jobGroup, jobDesc, false);
String imageWidth = getProperty("zeppelin.R.image.width", "100%");
@ -156,7 +155,15 @@ public class SparkRInterpreter extends Interpreter {
"\", \"" + jobDesc + "\", TRUE)";
}
lines = setJobGroup + "\n" + lines;
if (sparkInterpreter.getSparkVersion().newerThanEquals(SparkVersion.SPARK_2_3_0)) {
// setLocalProperty is only available from spark 2.3.0
String setPoolStmt = "setLocalProperty('spark.scheduler.pool', NULL)";
if (interpreterContext.getLocalProperties().containsKey("pool")) {
setPoolStmt = "setLocalProperty('spark.scheduler.pool', '" +
interpreterContext.getLocalProperties().get("pool") + "')";
}
lines = setPoolStmt + "\n" + lines;
}
try {
// render output with knitr
if (rbackendDead.get()) {

28
spark/interpreter/src/main/java/org/apache/zeppelin/spark/SparkSqlInterpreter.java
View file

@ -44,19 +44,13 @@ import org.slf4j.LoggerFactory;
public class SparkSqlInterpreter extends Interpreter {
private Logger logger = LoggerFactory.getLogger(SparkSqlInterpreter.class);
public static final String MAX_RESULTS = "zeppelin.spark.maxResult";
AtomicInteger num = new AtomicInteger(0);
private int maxResult;
public SparkSqlInterpreter(Properties property) {
super(property);
}
@Override
public void open() {
this.maxResult = Integer.parseInt(getProperty(MAX_RESULTS));
}
private SparkInterpreter getSparkInterpreter() throws InterpreterException {
@ -88,25 +82,17 @@ public class SparkSqlInterpreter extends Interpreter {
@Override
public InterpreterResult interpret(String st, InterpreterContext context)
throws InterpreterException {
SQLContext sqlc = null;
SparkInterpreter sparkInterpreter = getSparkInterpreter();
if (sparkInterpreter.isUnsupportedSparkVersion()) {
return new InterpreterResult(Code.ERROR, "Spark "
+ sparkInterpreter.getSparkVersion().toString() + " is not supported");
}
sparkInterpreter.getZeppelinContext().setInterpreterContext(context);
sqlc = sparkInterpreter.getSQLContext();
SQLContext sqlc = sparkInterpreter.getSQLContext();
SparkContext sc = sqlc.sparkContext();
if (concurrentSQL()) {
sc.setLocalProperty("spark.scheduler.pool", "fair");
} else {
sc.setLocalProperty("spark.scheduler.pool", null);
}
String jobDesc = "Started by: " + Utils.getUserName(context.getAuthenticationInfo());
sc.setJobGroup(Utils.buildJobGroupId(context), jobDesc, false);
sc.setLocalProperty("spark.scheduler.pool", context.getLocalProperties().get("pool"));
sc.setJobGroup(Utils.buildJobGroupId(context), Utils.buildJobDesc(context), false);
Object rdd = null;
try {
// method signature of sqlc.sql() is changed
@ -138,9 +124,7 @@ public class SparkSqlInterpreter extends Interpreter {
@Override
public void cancel(InterpreterContext context) throws InterpreterException {
SparkInterpreter sparkInterpreter = getSparkInterpreter();
SQLContext sqlc = sparkInterpreter.getSQLContext();
SparkContext sc = sqlc.sparkContext();
SparkContext sc = sparkInterpreter.getSparkContext();
sc.cancelJobGroup(Utils.buildJobGroupId(context));
}
@ -159,7 +143,7 @@ public class SparkSqlInterpreter extends Interpreter {
@Override
public Scheduler getScheduler() {
if (concurrentSQL()) {
int maxConcurrency = 10;
int maxConcurrency = Integer.parseInt(getProperty("zeppelin.spark.concurrentSQL", "10"));
return SchedulerFactory.singleton().createOrGetParallelScheduler(
SparkSqlInterpreter.class.getName() + this.hashCode(), maxConcurrency);
} else {

5
spark/interpreter/src/main/java/org/apache/zeppelin/spark/SparkVersion.java
View file

@ -34,6 +34,7 @@ public class SparkVersion {
public static final SparkVersion SPARK_1_6_0 = SparkVersion.fromVersionString("1.6.0");
public static final SparkVersion SPARK_2_0_0 = SparkVersion.fromVersionString("2.0.0");
public static final SparkVersion SPARK_2_3_0 = SparkVersion.fromVersionString("2.3.0");
public static final SparkVersion SPARK_2_3_1 = SparkVersion.fromVersionString("2.3.1");
public static final SparkVersion SPARK_2_4_0 = SparkVersion.fromVersionString("2.4.0");
@ -109,6 +110,10 @@ public class SparkVersion {
return this.olderThan(SPARK_1_3_0);
}
public boolean isSpark2() {
return this.newerThanEquals(SPARK_2_0_0);
}
public boolean isSecretSocketSupported() {
return this.newerThanEquals(SPARK_2_3_1);
}

3
spark/interpreter/src/main/java/org/apache/zeppelin/spark/SparkZeppelinContext.java
View file

@ -167,8 +167,7 @@ public class SparkZeppelinContext extends BaseZeppelinContext {
if (rows.length > maxResult) {
msg.append("\n");
msg.append(ResultMessages.getExceedsLimitRowsMessage(maxResult,
SparkSqlInterpreter.MAX_RESULTS));
msg.append(ResultMessages.getExceedsLimitRowsMessage(maxResult, "zeppelin.spark.maxResult"));
}
sc.clearJobGroup();

4
spark/interpreter/src/main/java/org/apache/zeppelin/spark/Utils.java
View file

@ -152,6 +152,10 @@ class Utils {
return "zeppelin-" + context.getNoteId() + "-" + context.getParagraphId();
}
public static String buildJobDesc(InterpreterContext context) {
return "Started by: " + getUserName(context.getAuthenticationInfo());
}
public static String getNoteId(String jobgroupId) {
int indexOf = jobgroupId.indexOf("-");
int secondIndex = jobgroupId.indexOf("-", indexOf + 1);

7
spark/interpreter/src/main/resources/interpreter-setting.json
View file

@ -102,6 +102,13 @@
"description": "Execute multiple SQL concurrently if set true.",
"type": "checkbox"
},
"zeppelin.spark.concurrentSQL.max": {
"envName": "ZEPPELIN_SPARK_CONCURRENTSQL_MAX",
"propertyName": "zeppelin.spark.concurrentSQL.max",
"defaultValue": 10,
"description": "Max number of SQL concurrently executed",
"type": "number"
},
"zeppelin.spark.sql.stacktrace": {
"envName": "ZEPPELIN_SPARK_SQL_STACKTRACE",
"propertyName": "zeppelin.spark.sql.stacktrace",

27
spark/interpreter/src/test/java/org/apache/zeppelin/spark/NewSparkInterpreterTest.java
View file

@ -423,6 +423,33 @@ public class NewSparkInterpreterTest {
assertEquals("hello world", output);
}
@Test
public void testSchedulePool() throws InterpreterException {
Properties properties = new Properties();
properties.setProperty("spark.master", "local");
properties.setProperty("spark.app.name", "test");
properties.setProperty("zeppelin.spark.maxResult", "100");
properties.setProperty("zeppelin.spark.test", "true");
properties.setProperty("zeppelin.spark.useNew", "true");
properties.setProperty("spark.scheduler.mode", "FAIR");
interpreter = new SparkInterpreter(properties);
assertTrue(interpreter.getDelegation() instanceof NewSparkInterpreter);
interpreter.setInterpreterGroup(mock(InterpreterGroup.class));
interpreter.open();
InterpreterContext context = getInterpreterContext();
context.getLocalProperties().put("pool", "pool1");
InterpreterResult result = interpreter.interpret("sc.range(1, 10).sum", context);
assertEquals(InterpreterResult.Code.SUCCESS, result.code());
assertEquals("pool1", interpreter.getSparkContext().getLocalProperty("spark.scheduler.pool"));
// pool is reset to null if user don't specify it via paragraph properties
result = interpreter.interpret("sc.range(1, 10).sum", getInterpreterContext());
assertEquals(InterpreterResult.Code.SUCCESS, result.code());
assertEquals(null, interpreter.getSparkContext().getLocalProperty("spark.scheduler.pool"));
}
@After
public void tearDown() throws InterpreterException {
if (this.interpreter != null) {

49
spark/interpreter/src/test/java/org/apache/zeppelin/spark/NewSparkSqlInterpreterTest.java
View file

@ -48,10 +48,10 @@ public class NewSparkSqlInterpreterTest {
@BeforeClass
public static void setUp() throws Exception {
Properties p = new Properties();
p.setProperty("spark.master", "local");
p.setProperty("spark.master", "local[4]");
p.setProperty("spark.app.name", "test");
p.setProperty("zeppelin.spark.maxResult", "10");
p.setProperty("zeppelin.spark.concurrentSQL", "false");
p.setProperty("zeppelin.spark.concurrentSQL", "true");
p.setProperty("zeppelin.spark.sqlInterpreter.stacktrace", "false");
p.setProperty("zeppelin.spark.useNew", "true");
intpGroup = new InterpreterGroup();
@ -179,4 +179,49 @@ public class NewSparkSqlInterpreterTest {
assertEquals(InterpreterResult.Code.SUCCESS, ret.code());
assertTrue(ret.message().get(1).getData().contains("alert-warning"));
}
@Test
public void testConcurrentSQL() throws InterpreterException, InterruptedException {
if (sparkInterpreter.getSparkVersion().isSpark2()) {
sparkInterpreter.interpret("spark.udf.register(\"sleep\", (e:Int) => {Thread.sleep(e*1000); e})", context);
} else {
sparkInterpreter.interpret("sqlContext.udf.register(\"sleep\", (e:Int) => {Thread.sleep(e*1000); e})", context);
}
Thread thread1 = new Thread() {
@Override
public void run() {
try {
InterpreterResult result = sqlInterpreter.interpret("select sleep(10)", context);
assertEquals(InterpreterResult.Code.SUCCESS, result.code());
} catch (InterpreterException e) {
e.printStackTrace();
}
}
};
Thread thread2 = new Thread() {
@Override
public void run() {
try {
InterpreterResult result = sqlInterpreter.interpret("select sleep(10)", context);
assertEquals(InterpreterResult.Code.SUCCESS, result.code());
} catch (InterpreterException e) {
e.printStackTrace();
}
}
};
// start running 2 spark sql, each would sleep 10 seconds, the totally running time should
// be less than 20 seconds, which means they run concurrently.
long start = System.currentTimeMillis();
thread1.start();
thread2.start();
thread1.join();
thread2.join();
long end = System.currentTimeMillis();
assertTrue("running time must be less than 20 seconds", ((end - start)/1000) < 20);
}
}

13
spark/interpreter/src/test/resources/fairscheduler.xml Normal file
View file

@ -0,0 +1,13 @@
<?xml version="1.0"?>
<allocations>
<pool name="pool1">
<schedulingMode>FAIR</schedulingMode>
<weight>1</weight>
<minShare>2</minShare>
</pool>
<pool name="pool2">
<schedulingMode>FIFO</schedulingMode>
<weight>2</weight>
<minShare>3</minShare>
</pool>
</allocations>

2
spark/spark-dependencies/pom.xml
View file

@ -44,7 +44,7 @@
instead of changing spark.version in this section.
-->
<hadoop.version>2.3.0</hadoop.version>
<hadoop.version>2.7.3</hadoop.version>
<yarn.version>${hadoop.version}</yarn.version>
<avro.version>1.7.7</avro.version>
<avro.mapred.classifier/>

24
zeppelin-interpreter/src/main/java/org/apache/zeppelin/conf/ZeppelinConfiguration.java
View file

@ -17,6 +17,12 @@
package org.apache.zeppelin.conf;
import java.io.File;
import java.net.URL;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import org.apache.commons.configuration.ConfigurationException;
import org.apache.commons.configuration.XMLConfiguration;
import org.apache.commons.configuration.tree.ConfigurationNode;
@ -25,13 +31,6 @@ import org.apache.zeppelin.util.Util;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.File;
import java.net.URL;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
/**
* Zeppelin configuration.
*
@ -494,10 +493,10 @@ public class ZeppelinConfiguration extends XMLConfiguration {
}
public String getRelativeDir(String path) {
if (path != null && path.startsWith("/") || isWindowsPath(path)) {
if (path != null && path.startsWith(File.separator) || isWindowsPath(path)) {
return path;
} else {
return getString(ConfVars.ZEPPELIN_HOME) + "/" + path;
return getString(ConfVars.ZEPPELIN_HOME) + File.separator + path;
}
}
@ -623,6 +622,10 @@ public class ZeppelinConfiguration extends XMLConfiguration {
return getString(ConfVars.ZEPPELIN_PROXY_PASSWORD);
}
public String getZeppelinSearchTempPath() {
return getRelativeDir(ConfVars.ZEPPELIN_SEARCH_TEMP_PATH);
}
public Map<String, String> dumpConfigurations(ZeppelinConfiguration conf,
ConfigurationKeyPredicate predicate) {
Map<String, String> configurations = new HashMap<>();
@ -823,7 +826,8 @@ public class ZeppelinConfiguration extends XMLConfiguration {
ZEPPELIN_NOTEBOOK_CRON_FOLDERS("zeppelin.notebook.cron.folders", null),
ZEPPELIN_PROXY_URL("zeppelin.proxy.url", null),
ZEPPELIN_PROXY_USER("zeppelin.proxy.user", null),
ZEPPELIN_PROXY_PASSWORD("zeppelin.proxy.password", null);
ZEPPELIN_PROXY_PASSWORD("zeppelin.proxy.password", null),
ZEPPELIN_SEARCH_TEMP_PATH("zeppelin.search.temp.path", System.getProperty("java.io.tmpdir"));
private String varName;
@SuppressWarnings("rawtypes")

10
zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/InterpreterContext.java
View file

@ -47,6 +47,7 @@ public class InterpreterContext {
}
private String noteId;
private String noteName;
private String replName;
private String paragraphTitle;
private String paragraphId;
@ -77,6 +78,11 @@ public class InterpreterContext {
return this;
}
public Builder setNoteName(String noteName) {
context.noteName = noteName;
return this;
}
public Builder setParagraphId(String paragraphId) {
context.paragraphId = paragraphId;
return this;
@ -171,6 +177,10 @@ public class InterpreterContext {
return noteId;
}
public String getNoteName() {
return noteName;
}
public String getReplName() {
return replName;
}

3
zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreterServer.java
View file

@ -751,6 +751,7 @@ public class RemoteInterpreterServer extends Thread
private InterpreterContext convert(RemoteInterpreterContext ric, InterpreterOutput output) {
return InterpreterContext.builder()
.setNoteId(ric.getNoteId())
.setNoteName(ric.getNoteName())
.setParagraphId(ric.getParagraphId())
.setReplName(ric.getReplName())
.setParagraphTitle(ric.getParagraphTitle())
@ -758,6 +759,8 @@ public class RemoteInterpreterServer extends Thread
.setLocalProperties(ric.getLocalProperties())
.setAuthenticationInfo(AuthenticationInfo.fromJson(ric.getAuthenticationInfo()))
.setGUI(GUI.fromJson(ric.getGui()))
.setConfig(gson.fromJson(ric.getConfig(),
new TypeToken<Map<String, Object>>() {}.getType()))
.setNoteGUI(GUI.fromJson(ric.getNoteGui()))
.setAngularObjectRegistry(interpreterGroup.getAngularObjectRegistry())
.setResourcePool(interpreterGroup.getResourcePool())

2
zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/AngularObjectId.java
View file

@ -51,7 +51,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@SuppressWarnings({"cast", "rawtypes", "serial", "unchecked"})
@Generated(value = "Autogenerated by Thrift Compiler (0.9.2)", date = "2018-6-21")
@Generated(value = "Autogenerated by Thrift Compiler (0.9.2)", date = "2018-7-6")
public class AngularObjectId implements org.apache.thrift.TBase<AngularObjectId, AngularObjectId._Fields>, java.io.Serializable, Cloneable, Comparable<AngularObjectId> {
private static final org.apache.thrift.protocol.TStruct STRUCT_DESC = new org.apache.thrift.protocol.TStruct("AngularObjectId");

2
zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/AppOutputAppendEvent.java
View file

@ -51,7 +51,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@SuppressWarnings({"cast", "rawtypes", "serial", "unchecked"})
@Generated(value = "Autogenerated by Thrift Compiler (0.9.2)", date = "2018-6-21")
@Generated(value = "Autogenerated by Thrift Compiler (0.9.2)", date = "2018-7-6")
public class AppOutputAppendEvent implements org.apache.thrift.TBase<AppOutputAppendEvent, AppOutputAppendEvent._Fields>, java.io.Serializable, Cloneable, Comparable<AppOutputAppendEvent> {
private static final org.apache.thrift.protocol.TStruct STRUCT_DESC = new org.apache.thrift.protocol.TStruct("AppOutputAppendEvent");

2
zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/AppOutputUpdateEvent.java
View file

@ -51,7 +51,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@SuppressWarnings({"cast", "rawtypes", "serial", "unchecked"})
@Generated(value = "Autogenerated by Thrift Compiler (0.9.2)", date = "2018-6-21")
@Generated(value = "Autogenerated by Thrift Compiler (0.9.2)", date = "2018-7-6")
public class AppOutputUpdateEvent implements org.apache.thrift.TBase<AppOutputUpdateEvent, AppOutputUpdateEvent._Fields>, java.io.Serializable, Cloneable, Comparable<AppOutputUpdateEvent> {
private static final org.apache.thrift.protocol.TStruct STRUCT_DESC = new org.apache.thrift.protocol.TStruct("AppOutputUpdateEvent");

2
zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/AppStatusUpdateEvent.java
View file

@ -51,7 +51,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@SuppressWarnings({"cast", "rawtypes", "serial", "unchecked"})
@Generated(value = "Autogenerated by Thrift Compiler (0.9.2)", date = "2018-6-21")
@Generated(value = "Autogenerated by Thrift Compiler (0.9.2)", date = "2018-7-6")
public class AppStatusUpdateEvent implements org.apache.thrift.TBase<AppStatusUpdateEvent, AppStatusUpdateEvent._Fields>, java.io.Serializable, Cloneable, Comparable<AppStatusUpdateEvent> {
private static final org.apache.thrift.protocol.TStruct STRUCT_DESC = new org.apache.thrift.protocol.TStruct("AppStatusUpdateEvent");

2
zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/InterpreterCompletion.java
View file

@ -51,7 +51,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@SuppressWarnings({"cast", "rawtypes", "serial", "unchecked"})
@Generated(value = "Autogenerated by Thrift Compiler (0.9.2)", date = "2018-6-21")
@Generated(value = "Autogenerated by Thrift Compiler (0.9.2)", date = "2018-7-6")
public class InterpreterCompletion implements org.apache.thrift.TBase<InterpreterCompletion, InterpreterCompletion._Fields>, java.io.Serializable, Cloneable, Comparable<InterpreterCompletion> {
private static final org.apache.thrift.protocol.TStruct STRUCT_DESC = new org.apache.thrift.protocol.TStruct("InterpreterCompletion");

2
zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/OutputAppendEvent.java
View file

@ -51,7 +51,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@SuppressWarnings({"cast", "rawtypes", "serial", "unchecked"})
@Generated(value = "Autogenerated by Thrift Compiler (0.9.2)", date = "2018-6-21")
@Generated(value = "Autogenerated by Thrift Compiler (0.9.2)", date = "2018-7-6")
public class OutputAppendEvent implements org.apache.thrift.TBase<OutputAppendEvent, OutputAppendEvent._Fields>, java.io.Serializable, Cloneable, Comparable<OutputAppendEvent> {
private static final org.apache.thrift.protocol.TStruct STRUCT_DESC = new org.apache.thrift.protocol.TStruct("OutputAppendEvent");

2
zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/OutputUpdateAllEvent.java
View file

@ -51,7 +51,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@SuppressWarnings({"cast", "rawtypes", "serial", "unchecked"})
@Generated(value = "Autogenerated by Thrift Compiler (0.9.2)", date = "2018-6-21")
@Generated(value = "Autogenerated by Thrift Compiler (0.9.2)", date = "2018-7-6")
public class OutputUpdateAllEvent implements org.apache.thrift.TBase<OutputUpdateAllEvent, OutputUpdateAllEvent._Fields>, java.io.Serializable, Cloneable, Comparable<OutputUpdateAllEvent> {
private static final org.apache.thrift.protocol.TStruct STRUCT_DESC = new org.apache.thrift.protocol.TStruct("OutputUpdateAllEvent");

2
zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/OutputUpdateEvent.java
View file

@ -51,7 +51,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@SuppressWarnings({"cast", "rawtypes", "serial", "unchecked"})
@Generated(value = "Autogenerated by Thrift Compiler (0.9.2)", date = "2018-6-21")
@Generated(value = "Autogenerated by Thrift Compiler (0.9.2)", date = "2018-7-6")
public class OutputUpdateEvent implements org.apache.thrift.TBase<OutputUpdateEvent, OutputUpdateEvent._Fields>, java.io.Serializable, Cloneable, Comparable<OutputUpdateEvent> {
private static final org.apache.thrift.protocol.TStruct STRUCT_DESC = new org.apache.thrift.protocol.TStruct("OutputUpdateEvent");

2
zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/RegisterInfo.java
View file

@ -51,7 +51,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@SuppressWarnings({"cast", "rawtypes", "serial", "unchecked"})
@Generated(value = "Autogenerated by Thrift Compiler (0.9.2)", date = "2018-6-21")
@Generated(value = "Autogenerated by Thrift Compiler (0.9.2)", date = "2018-7-6")
public class RegisterInfo implements org.apache.thrift.TBase<RegisterInfo, RegisterInfo._Fields>, java.io.Serializable, Cloneable, Comparable<RegisterInfo> {
private static final org.apache.thrift.protocol.TStruct STRUCT_DESC = new org.apache.thrift.protocol.TStruct("RegisterInfo");

2
zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/RemoteApplicationResult.java
View file

@ -51,7 +51,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@SuppressWarnings({"cast", "rawtypes", "serial", "unchecked"})
@Generated(value = "Autogenerated by Thrift Compiler (0.9.2)", date = "2018-6-21")
@Generated(value = "Autogenerated by Thrift Compiler (0.9.2)", date = "2018-7-6")
public class RemoteApplicationResult implements org.apache.thrift.TBase<RemoteApplicationResult, RemoteApplicationResult._Fields>, java.io.Serializable, Cloneable, Comparable<RemoteApplicationResult> {
private static final org.apache.thrift.protocol.TStruct STRUCT_DESC = new org.apache.thrift.protocol.TStruct("RemoteApplicationResult");

217
zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/RemoteInterpreterContext.java
View file

@ -51,20 +51,21 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@SuppressWarnings({"cast", "rawtypes", "serial", "unchecked"})
@Generated(value = "Autogenerated by Thrift Compiler (0.9.2)", date = "2018-6-21")
@Generated(value = "Autogenerated by Thrift Compiler (0.9.2)", date = "2018-7-6")
public class RemoteInterpreterContext implements org.apache.thrift.TBase<RemoteInterpreterContext, RemoteInterpreterContext._Fields>, java.io.Serializable, Cloneable, Comparable<RemoteInterpreterContext> {
private static final org.apache.thrift.protocol.TStruct STRUCT_DESC = new org.apache.thrift.protocol.TStruct("RemoteInterpreterContext");
private static final org.apache.thrift.protocol.TField NOTE_ID_FIELD_DESC = new org.apache.thrift.protocol.TField("noteId", org.apache.thrift.protocol.TType.STRING, (short)1);
private static final org.apache.thrift.protocol.TField PARAGRAPH_ID_FIELD_DESC = new org.apache.thrift.protocol.TField("paragraphId", org.apache.thrift.protocol.TType.STRING, (short)2);
private static final org.apache.thrift.protocol.TField REPL_NAME_FIELD_DESC = new org.apache.thrift.protocol.TField("replName", org.apache.thrift.protocol.TType.STRING, (short)3);
private static final org.apache.thrift.protocol.TField PARAGRAPH_TITLE_FIELD_DESC = new org.apache.thrift.protocol.TField("paragraphTitle", org.apache.thrift.protocol.TType.STRING, (short)4);
private static final org.apache.thrift.protocol.TField PARAGRAPH_TEXT_FIELD_DESC = new org.apache.thrift.protocol.TField("paragraphText", org.apache.thrift.protocol.TType.STRING, (short)5);
private static final org.apache.thrift.protocol.TField AUTHENTICATION_INFO_FIELD_DESC = new org.apache.thrift.protocol.TField("authenticationInfo", org.apache.thrift.protocol.TType.STRING, (short)6);
private static final org.apache.thrift.protocol.TField CONFIG_FIELD_DESC = new org.apache.thrift.protocol.TField("config", org.apache.thrift.protocol.TType.STRING, (short)7);
private static final org.apache.thrift.protocol.TField GUI_FIELD_DESC = new org.apache.thrift.protocol.TField("gui", org.apache.thrift.protocol.TType.STRING, (short)8);
private static final org.apache.thrift.protocol.TField NOTE_GUI_FIELD_DESC = new org.apache.thrift.protocol.TField("noteGui", org.apache.thrift.protocol.TType.STRING, (short)9);
private static final org.apache.thrift.protocol.TField LOCAL_PROPERTIES_FIELD_DESC = new org.apache.thrift.protocol.TField("localProperties", org.apache.thrift.protocol.TType.MAP, (short)10);
private static final org.apache.thrift.protocol.TField NOTE_NAME_FIELD_DESC = new org.apache.thrift.protocol.TField("noteName", org.apache.thrift.protocol.TType.STRING, (short)2);
private static final org.apache.thrift.protocol.TField PARAGRAPH_ID_FIELD_DESC = new org.apache.thrift.protocol.TField("paragraphId", org.apache.thrift.protocol.TType.STRING, (short)3);
private static final org.apache.thrift.protocol.TField REPL_NAME_FIELD_DESC = new org.apache.thrift.protocol.TField("replName", org.apache.thrift.protocol.TType.STRING, (short)4);
private static final org.apache.thrift.protocol.TField PARAGRAPH_TITLE_FIELD_DESC = new org.apache.thrift.protocol.TField("paragraphTitle", org.apache.thrift.protocol.TType.STRING, (short)5);
private static final org.apache.thrift.protocol.TField PARAGRAPH_TEXT_FIELD_DESC = new org.apache.thrift.protocol.TField("paragraphText", org.apache.thrift.protocol.TType.STRING, (short)6);
private static final org.apache.thrift.protocol.TField AUTHENTICATION_INFO_FIELD_DESC = new org.apache.thrift.protocol.TField("authenticationInfo", org.apache.thrift.protocol.TType.STRING, (short)7);
private static final org.apache.thrift.protocol.TField CONFIG_FIELD_DESC = new org.apache.thrift.protocol.TField("config", org.apache.thrift.protocol.TType.STRING, (short)8);
private static final org.apache.thrift.protocol.TField GUI_FIELD_DESC = new org.apache.thrift.protocol.TField("gui", org.apache.thrift.protocol.TType.STRING, (short)9);
private static final org.apache.thrift.protocol.TField NOTE_GUI_FIELD_DESC = new org.apache.thrift.protocol.TField("noteGui", org.apache.thrift.protocol.TType.STRING, (short)10);
private static final org.apache.thrift.protocol.TField LOCAL_PROPERTIES_FIELD_DESC = new org.apache.thrift.protocol.TField("localProperties", org.apache.thrift.protocol.TType.MAP, (short)11);
private static final Map<Class<? extends IScheme>, SchemeFactory> schemes = new HashMap<Class<? extends IScheme>, SchemeFactory>();
static {
@ -73,6 +74,7 @@ public class RemoteInterpreterContext implements org.apache.thrift.TBase<RemoteI
}
public String noteId; // required
public String noteName; // required
public String paragraphId; // required
public String replName; // required
public String paragraphTitle; // required
@ -86,15 +88,16 @@ public class RemoteInterpreterContext implements org.apache.thrift.TBase<RemoteI
/** The set of fields this struct contains, along with convenience methods for finding and manipulating them. */
public enum _Fields implements org.apache.thrift.TFieldIdEnum {
NOTE_ID((short)1, "noteId"),
PARAGRAPH_ID((short)2, "paragraphId"),
REPL_NAME((short)3, "replName"),
PARAGRAPH_TITLE((short)4, "paragraphTitle"),
PARAGRAPH_TEXT((short)5, "paragraphText"),
AUTHENTICATION_INFO((short)6, "authenticationInfo"),
CONFIG((short)7, "config"),
GUI((short)8, "gui"),
NOTE_GUI((short)9, "noteGui"),
LOCAL_PROPERTIES((short)10, "localProperties");
NOTE_NAME((short)2, "noteName"),
PARAGRAPH_ID((short)3, "paragraphId"),
REPL_NAME((short)4, "replName"),
PARAGRAPH_TITLE((short)5, "paragraphTitle"),
PARAGRAPH_TEXT((short)6, "paragraphText"),
AUTHENTICATION_INFO((short)7, "authenticationInfo"),
CONFIG((short)8, "config"),
GUI((short)9, "gui"),
NOTE_GUI((short)10, "noteGui"),
LOCAL_PROPERTIES((short)11, "localProperties");
private static final Map<String, _Fields> byName = new HashMap<String, _Fields>();
@ -111,23 +114,25 @@ public class RemoteInterpreterContext implements org.apache.thrift.TBase<RemoteI
switch(fieldId) {
case 1: // NOTE_ID
return NOTE_ID;
case 2: // PARAGRAPH_ID
case 2: // NOTE_NAME
return NOTE_NAME;
case 3: // PARAGRAPH_ID
return PARAGRAPH_ID;
case 3: // REPL_NAME
case 4: // REPL_NAME
return REPL_NAME;
case 4: // PARAGRAPH_TITLE
case 5: // PARAGRAPH_TITLE
return PARAGRAPH_TITLE;
case 5: // PARAGRAPH_TEXT
case 6: // PARAGRAPH_TEXT
return PARAGRAPH_TEXT;
case 6: // AUTHENTICATION_INFO
case 7: // AUTHENTICATION_INFO
return AUTHENTICATION_INFO;
case 7: // CONFIG
case 8: // CONFIG
return CONFIG;
case 8: // GUI
case 9: // GUI
return GUI;
case 9: // NOTE_GUI
case 10: // NOTE_GUI
return NOTE_GUI;
case 10: // LOCAL_PROPERTIES
case 11: // LOCAL_PROPERTIES
return LOCAL_PROPERTIES;
default:
return null;
@ -174,6 +179,8 @@ public class RemoteInterpreterContext implements org.apache.thrift.TBase<RemoteI
Map<_Fields, org.apache.thrift.meta_data.FieldMetaData> tmpMap = new EnumMap<_Fields, org.apache.thrift.meta_data.FieldMetaData>(_Fields.class);
tmpMap.put(_Fields.NOTE_ID, new org.apache.thrift.meta_data.FieldMetaData("noteId", org.apache.thrift.TFieldRequirementType.DEFAULT,
new org.apache.thrift.meta_data.FieldValueMetaData(org.apache.thrift.protocol.TType.STRING)));
tmpMap.put(_Fields.NOTE_NAME, new org.apache.thrift.meta_data.FieldMetaData("noteName", org.apache.thrift.TFieldRequirementType.DEFAULT,
new org.apache.thrift.meta_data.FieldValueMetaData(org.apache.thrift.protocol.TType.STRING)));
tmpMap.put(_Fields.PARAGRAPH_ID, new org.apache.thrift.meta_data.FieldMetaData("paragraphId", org.apache.thrift.TFieldRequirementType.DEFAULT,
new org.apache.thrift.meta_data.FieldValueMetaData(org.apache.thrift.protocol.TType.STRING)));
tmpMap.put(_Fields.REPL_NAME, new org.apache.thrift.meta_data.FieldMetaData("replName", org.apache.thrift.TFieldRequirementType.DEFAULT,
@ -203,6 +210,7 @@ public class RemoteInterpreterContext implements org.apache.thrift.TBase<RemoteI
public RemoteInterpreterContext(
String noteId,
String noteName,
String paragraphId,
String replName,
String paragraphTitle,
@ -215,6 +223,7 @@ public class RemoteInterpreterContext implements org.apache.thrift.TBase<RemoteI
{
this();
this.noteId = noteId;
this.noteName = noteName;
this.paragraphId = paragraphId;
this.replName = replName;
this.paragraphTitle = paragraphTitle;
@ -233,6 +242,9 @@ public class RemoteInterpreterContext implements org.apache.thrift.TBase<RemoteI
if (other.isSetNoteId()) {
this.noteId = other.noteId;
}
if (other.isSetNoteName()) {
this.noteName = other.noteName;
}
if (other.isSetParagraphId()) {
this.paragraphId = other.paragraphId;
}
@ -270,6 +282,7 @@ public class RemoteInterpreterContext implements org.apache.thrift.TBase<RemoteI
@Override
public void clear() {
this.noteId = null;
this.noteName = null;
this.paragraphId = null;
this.replName = null;
this.paragraphTitle = null;
@ -305,6 +318,30 @@ public class RemoteInterpreterContext implements org.apache.thrift.TBase<RemoteI
}
}
public String getNoteName() {
return this.noteName;
}
public RemoteInterpreterContext setNoteName(String noteName) {
this.noteName = noteName;
return this;
}
public void unsetNoteName() {
this.noteName = null;
}
/** Returns true if field noteName is set (has been assigned a value) and false otherwise */
public boolean isSetNoteName() {
return this.noteName != null;
}
public void setNoteNameIsSet(boolean value) {
if (!value) {
this.noteName = null;
}
}
public String getParagraphId() {
return this.paragraphId;
}
@ -542,6 +579,14 @@ public class RemoteInterpreterContext implements org.apache.thrift.TBase<RemoteI
}
break;
case NOTE_NAME:
if (value == null) {
unsetNoteName();
} else {
setNoteName((String)value);
}
break;
case PARAGRAPH_ID:
if (value == null) {
unsetParagraphId();
@ -622,6 +667,9 @@ public class RemoteInterpreterContext implements org.apache.thrift.TBase<RemoteI
case NOTE_ID:
return getNoteId();
case NOTE_NAME:
return getNoteName();
case PARAGRAPH_ID:
return getParagraphId();
@ -662,6 +710,8 @@ public class RemoteInterpreterContext implements org.apache.thrift.TBase<RemoteI
switch (field) {
case NOTE_ID:
return isSetNoteId();
case NOTE_NAME:
return isSetNoteName();
case PARAGRAPH_ID:
return isSetParagraphId();
case REPL_NAME:
@ -706,6 +756,15 @@ public class RemoteInterpreterContext implements org.apache.thrift.TBase<RemoteI
return false;
}
boolean this_present_noteName = true && this.isSetNoteName();
boolean that_present_noteName = true && that.isSetNoteName();
if (this_present_noteName || that_present_noteName) {
if (!(this_present_noteName && that_present_noteName))
return false;
if (!this.noteName.equals(that.noteName))
return false;
}
boolean this_present_paragraphId = true && this.isSetParagraphId();
boolean that_present_paragraphId = true && that.isSetParagraphId();
if (this_present_paragraphId || that_present_paragraphId) {
@ -799,6 +858,11 @@ public class RemoteInterpreterContext implements org.apache.thrift.TBase<RemoteI
if (present_noteId)
list.add(noteId);
boolean present_noteName = true && (isSetNoteName());
list.add(present_noteName);
if (present_noteName)
list.add(noteName);
boolean present_paragraphId = true && (isSetParagraphId());
list.add(present_paragraphId);
if (present_paragraphId)
@ -865,6 +929,16 @@ public class RemoteInterpreterContext implements org.apache.thrift.TBase<RemoteI
return lastComparison;
}
}
lastComparison = Boolean.valueOf(isSetNoteName()).compareTo(other.isSetNoteName());
if (lastComparison != 0) {
return lastComparison;
}
if (isSetNoteName()) {
lastComparison = org.apache.thrift.TBaseHelper.compareTo(this.noteName, other.noteName);
if (lastComparison != 0) {
return lastComparison;
}
}
lastComparison = Boolean.valueOf(isSetParagraphId()).compareTo(other.isSetParagraphId());
if (lastComparison != 0) {
return lastComparison;
@ -983,6 +1057,14 @@ public class RemoteInterpreterContext implements org.apache.thrift.TBase<RemoteI
}
first = false;
if (!first) sb.append(", ");
sb.append("noteName:");
if (this.noteName == null) {
sb.append("null");
} else {
sb.append(this.noteName);
}
first = false;
if (!first) sb.append(", ");
sb.append("paragraphId:");
if (this.paragraphId == null) {
sb.append("null");
@ -1105,7 +1187,15 @@ public class RemoteInterpreterContext implements org.apache.thrift.TBase<RemoteI
org.apache.thrift.protocol.TProtocolUtil.skip(iprot, schemeField.type);
}
break;
case 2: // PARAGRAPH_ID
case 2: // NOTE_NAME
if (schemeField.type == org.apache.thrift.protocol.TType.STRING) {
struct.noteName = iprot.readString();
struct.setNoteNameIsSet(true);
} else {
org.apache.thrift.protocol.TProtocolUtil.skip(iprot, schemeField.type);
}
break;
case 3: // PARAGRAPH_ID
if (schemeField.type == org.apache.thrift.protocol.TType.STRING) {
struct.paragraphId = iprot.readString();
struct.setParagraphIdIsSet(true);
@ -1113,7 +1203,7 @@ public class RemoteInterpreterContext implements org.apache.thrift.TBase<RemoteI
org.apache.thrift.protocol.TProtocolUtil.skip(iprot, schemeField.type);
}
break;
case 3: // REPL_NAME
case 4: // REPL_NAME
if (schemeField.type == org.apache.thrift.protocol.TType.STRING) {
struct.replName = iprot.readString();
struct.setReplNameIsSet(true);
@ -1121,7 +1211,7 @@ public class RemoteInterpreterContext implements org.apache.thrift.TBase<RemoteI
org.apache.thrift.protocol.TProtocolUtil.skip(iprot, schemeField.type);
}
break;
case 4: // PARAGRAPH_TITLE
case 5: // PARAGRAPH_TITLE
if (schemeField.type == org.apache.thrift.protocol.TType.STRING) {
struct.paragraphTitle = iprot.readString();
struct.setParagraphTitleIsSet(true);
@ -1129,7 +1219,7 @@ public class RemoteInterpreterContext implements org.apache.thrift.TBase<RemoteI
org.apache.thrift.protocol.TProtocolUtil.skip(iprot, schemeField.type);
}
break;
case 5: // PARAGRAPH_TEXT
case 6: // PARAGRAPH_TEXT
if (schemeField.type == org.apache.thrift.protocol.TType.STRING) {
struct.paragraphText = iprot.readString();
struct.setParagraphTextIsSet(true);
@ -1137,7 +1227,7 @@ public class RemoteInterpreterContext implements org.apache.thrift.TBase<RemoteI
org.apache.thrift.protocol.TProtocolUtil.skip(iprot, schemeField.type);
}
break;
case 6: // AUTHENTICATION_INFO
case 7: // AUTHENTICATION_INFO
if (schemeField.type == org.apache.thrift.protocol.TType.STRING) {
struct.authenticationInfo = iprot.readString();
struct.setAuthenticationInfoIsSet(true);
@ -1145,7 +1235,7 @@ public class RemoteInterpreterContext implements org.apache.thrift.TBase<RemoteI
org.apache.thrift.protocol.TProtocolUtil.skip(iprot, schemeField.type);
}
break;
case 7: // CONFIG
case 8: // CONFIG
if (schemeField.type == org.apache.thrift.protocol.TType.STRING) {
struct.config = iprot.readString();
struct.setConfigIsSet(true);
@ -1153,7 +1243,7 @@ public class RemoteInterpreterContext implements org.apache.thrift.TBase<RemoteI
org.apache.thrift.protocol.TProtocolUtil.skip(iprot, schemeField.type);
}
break;
case 8: // GUI
case 9: // GUI
if (schemeField.type == org.apache.thrift.protocol.TType.STRING) {
struct.gui = iprot.readString();
struct.setGuiIsSet(true);
@ -1161,7 +1251,7 @@ public class RemoteInterpreterContext implements org.apache.thrift.TBase<RemoteI
org.apache.thrift.protocol.TProtocolUtil.skip(iprot, schemeField.type);
}
break;
case 9: // NOTE_GUI
case 10: // NOTE_GUI
if (schemeField.type == org.apache.thrift.protocol.TType.STRING) {
struct.noteGui = iprot.readString();
struct.setNoteGuiIsSet(true);
@ -1169,7 +1259,7 @@ public class RemoteInterpreterContext implements org.apache.thrift.TBase<RemoteI
org.apache.thrift.protocol.TProtocolUtil.skip(iprot, schemeField.type);
}
break;
case 10: // LOCAL_PROPERTIES
case 11: // LOCAL_PROPERTIES
if (schemeField.type == org.apache.thrift.protocol.TType.MAP) {
{
org.apache.thrift.protocol.TMap _map0 = iprot.readMapBegin();
@ -1209,6 +1299,11 @@ public class RemoteInterpreterContext implements org.apache.thrift.TBase<RemoteI
oprot.writeString(struct.noteId);
oprot.writeFieldEnd();
}
if (struct.noteName != null) {
oprot.writeFieldBegin(NOTE_NAME_FIELD_DESC);
oprot.writeString(struct.noteName);
oprot.writeFieldEnd();
}
if (struct.paragraphId != null) {
oprot.writeFieldBegin(PARAGRAPH_ID_FIELD_DESC);
oprot.writeString(struct.paragraphId);
@ -1283,37 +1378,43 @@ public class RemoteInterpreterContext implements org.apache.thrift.TBase<RemoteI
if (struct.isSetNoteId()) {
optionals.set(0);
}
if (struct.isSetParagraphId()) {
if (struct.isSetNoteName()) {
optionals.set(1);
}
if (struct.isSetReplName()) {
if (struct.isSetParagraphId()) {
optionals.set(2);
}
if (struct.isSetParagraphTitle()) {
if (struct.isSetReplName()) {
optionals.set(3);
}
if (struct.isSetParagraphText()) {
if (struct.isSetParagraphTitle()) {
optionals.set(4);
}
if (struct.isSetAuthenticationInfo()) {
if (struct.isSetParagraphText()) {
optionals.set(5);
}
if (struct.isSetConfig()) {
if (struct.isSetAuthenticationInfo()) {
optionals.set(6);
}
if (struct.isSetGui()) {
if (struct.isSetConfig()) {
optionals.set(7);
}
if (struct.isSetNoteGui()) {
if (struct.isSetGui()) {
optionals.set(8);
}
if (struct.isSetLocalProperties()) {
if (struct.isSetNoteGui()) {
optionals.set(9);
}
oprot.writeBitSet(optionals, 10);
if (struct.isSetLocalProperties()) {
optionals.set(10);
}
oprot.writeBitSet(optionals, 11);
if (struct.isSetNoteId()) {
oprot.writeString(struct.noteId);
}
if (struct.isSetNoteName()) {
oprot.writeString(struct.noteName);
}
if (struct.isSetParagraphId()) {
oprot.writeString(struct.paragraphId);
}
@ -1353,44 +1454,48 @@ public class RemoteInterpreterContext implements org.apache.thrift.TBase<RemoteI
@Override
public void read(org.apache.thrift.protocol.TProtocol prot, RemoteInterpreterContext struct) throws org.apache.thrift.TException {
TTupleProtocol iprot = (TTupleProtocol) prot;
BitSet incoming = iprot.readBitSet(10);
BitSet incoming = iprot.readBitSet(11);
if (incoming.get(0)) {
struct.noteId = iprot.readString();
struct.setNoteIdIsSet(true);
}
if (incoming.get(1)) {
struct.noteName = iprot.readString();
struct.setNoteNameIsSet(true);
}
if (incoming.get(2)) {
struct.paragraphId = iprot.readString();
struct.setParagraphIdIsSet(true);
}
if (incoming.get(2)) {
if (incoming.get(3)) {
struct.replName = iprot.readString();
struct.setReplNameIsSet(true);
}
if (incoming.get(3)) {
if (incoming.get(4)) {
struct.paragraphTitle = iprot.readString();
struct.setParagraphTitleIsSet(true);
}
if (incoming.get(4)) {
if (incoming.get(5)) {
struct.paragraphText = iprot.readString();
struct.setParagraphTextIsSet(true);
}
if (incoming.get(5)) {
if (incoming.get(6)) {
struct.authenticationInfo = iprot.readString();
struct.setAuthenticationInfoIsSet(true);
}
if (incoming.get(6)) {
if (incoming.get(7)) {
struct.config = iprot.readString();
struct.setConfigIsSet(true);
}
if (incoming.get(7)) {
if (incoming.get(8)) {
struct.gui = iprot.readString();
struct.setGuiIsSet(true);
}
if (incoming.get(8)) {
if (incoming.get(9)) {
struct.noteGui = iprot.readString();
struct.setNoteGuiIsSet(true);
}
if (incoming.get(9)) {
if (incoming.get(10)) {
{
org.apache.thrift.protocol.TMap _map6 = new org.apache.thrift.protocol.TMap(org.apache.thrift.protocol.TType.STRING, org.apache.thrift.protocol.TType.STRING, iprot.readI32());
struct.localProperties = new HashMap<String,String>(2*_map6.size);

2
zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/RemoteInterpreterEvent.java
View file

@ -51,7 +51,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@SuppressWarnings({"cast", "rawtypes", "serial", "unchecked"})
@Generated(value = "Autogenerated by Thrift Compiler (0.9.2)", date = "2018-6-21")
@Generated(value = "Autogenerated by Thrift Compiler (0.9.2)", date = "2018-7-6")
public class RemoteInterpreterEvent implements org.apache.thrift.TBase<RemoteInterpreterEvent, RemoteInterpreterEvent._Fields>, java.io.Serializable, Cloneable, Comparable<RemoteInterpreterEvent> {
private static final org.apache.thrift.protocol.TStruct STRUCT_DESC = new org.apache.thrift.protocol.TStruct("RemoteInterpreterEvent");

2
zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/RemoteInterpreterEventService.java
View file

@ -51,7 +51,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@SuppressWarnings({"cast", "rawtypes", "serial", "unchecked"})
@Generated(value = "Autogenerated by Thrift Compiler (0.9.2)", date = "2018-6-21")
@Generated(value = "Autogenerated by Thrift Compiler (0.9.2)", date = "2018-7-6")
public class RemoteInterpreterEventService {
public interface Iface {

2
zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/RemoteInterpreterResult.java
View file

@ -51,7 +51,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@SuppressWarnings({"cast", "rawtypes", "serial", "unchecked"})
@Generated(value = "Autogenerated by Thrift Compiler (0.9.2)", date = "2018-6-21")
@Generated(value = "Autogenerated by Thrift Compiler (0.9.2)", date = "2018-7-6")
public class RemoteInterpreterResult implements org.apache.thrift.TBase<RemoteInterpreterResult, RemoteInterpreterResult._Fields>, java.io.Serializable, Cloneable, Comparable<RemoteInterpreterResult> {
private static final org.apache.thrift.protocol.TStruct STRUCT_DESC = new org.apache.thrift.protocol.TStruct("RemoteInterpreterResult");

2
zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/RemoteInterpreterResultMessage.java
View file

@ -51,7 +51,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@SuppressWarnings({"cast", "rawtypes", "serial", "unchecked"})
@Generated(value = "Autogenerated by Thrift Compiler (0.9.2)", date = "2018-6-21")
@Generated(value = "Autogenerated by Thrift Compiler (0.9.2)", date = "2018-7-6")
public class RemoteInterpreterResultMessage implements org.apache.thrift.TBase<RemoteInterpreterResultMessage, RemoteInterpreterResultMessage._Fields>, java.io.Serializable, Cloneable, Comparable<RemoteInterpreterResultMessage> {
private static final org.apache.thrift.protocol.TStruct STRUCT_DESC = new org.apache.thrift.protocol.TStruct("RemoteInterpreterResultMessage");

2
zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/RemoteInterpreterService.java
View file

@ -51,7 +51,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@SuppressWarnings({"cast", "rawtypes", "serial", "unchecked"})
@Generated(value = "Autogenerated by Thrift Compiler (0.9.2)", date = "2018-6-21")
@Generated(value = "Autogenerated by Thrift Compiler (0.9.2)", date = "2018-7-6")
public class RemoteInterpreterService {
public interface Iface {

2
zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/RunParagraphsEvent.java
View file

@ -51,7 +51,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@SuppressWarnings({"cast", "rawtypes", "serial", "unchecked"})
@Generated(value = "Autogenerated by Thrift Compiler (0.9.2)", date = "2018-6-21")
@Generated(value = "Autogenerated by Thrift Compiler (0.9.2)", date = "2018-7-6")
public class RunParagraphsEvent implements org.apache.thrift.TBase<RunParagraphsEvent, RunParagraphsEvent._Fields>, java.io.Serializable, Cloneable, Comparable<RunParagraphsEvent> {
private static final org.apache.thrift.protocol.TStruct STRUCT_DESC = new org.apache.thrift.protocol.TStruct("RunParagraphsEvent");

2
zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/ZeppelinServerResourceParagraphRunner.java
View file

@ -51,7 +51,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@SuppressWarnings({"cast", "rawtypes", "serial", "unchecked"})
@Generated(value = "Autogenerated by Thrift Compiler (0.9.2)", date = "2018-6-21")
@Generated(value = "Autogenerated by Thrift Compiler (0.9.2)", date = "2018-7-6")
public class ZeppelinServerResourceParagraphRunner implements org.apache.thrift.TBase<ZeppelinServerResourceParagraphRunner, ZeppelinServerResourceParagraphRunner._Fields>, java.io.Serializable, Cloneable, Comparable<ZeppelinServerResourceParagraphRunner> {
private static final org.apache.thrift.protocol.TStruct STRUCT_DESC = new org.apache.thrift.protocol.TStruct("ZeppelinServerResourceParagraphRunner");

19
zeppelin-interpreter/src/main/thrift/RemoteInterpreterService.thrift
View file

@ -20,15 +20,16 @@ namespace java org.apache.zeppelin.interpreter.thrift
struct RemoteInterpreterContext {
1: string noteId,
2: string paragraphId,
3: string replName,
4: string paragraphTitle,
5: string paragraphText,
6: string authenticationInfo,
7: string config, // json serialized config
8: string gui, // json serialized gui
9: string noteGui, // json serialized note gui
10: map<string, string> localProperties
2: string noteName,
3: string paragraphId,
4: string replName,
5: string paragraphTitle,
6: string paragraphText,
7: string authenticationInfo,
8: string config, // json serialized config
9: string gui, // json serialized gui
10: string noteGui, // json serialized note gui
11: map<string, string> localProperties
}
struct RemoteInterpreterResultMessage {

9
zeppelin-interpreter/src/test/java/org/apache/zeppelin/interpreter/BaseZeppelinContextTest.java
View file

@ -34,12 +34,21 @@ public class BaseZeppelinContextTest {
TestZeppelinContext z = new TestZeppelinContext(hookRegistry, 10);
InterpreterContext context = InterpreterContext.builder()
.setNoteId("note_1")
.setNoteName("note_name_1")
.setParagraphId("paragraph_1")
.setInterpreterClassName("Test1Interpreter")
.setReplName("test1")
.build();
z.setInterpreterContext(context);
// get note name via InterpreterContext
String note_name = z.getInterpreterContext().getNoteName();
assertEquals(
String.format("Actual note name: %s, but expected %s", note_name, "note_name_1"),
"note_name_1",
note_name
);
// register global hook for current interpreter
z.registerHook(InterpreterHookRegistry.HookType.PRE_EXEC.getName(), "pre_cmd");
z.registerHook(InterpreterHookRegistry.HookType.POST_EXEC.getName(), "post_cmd");

24
zeppelin-plugins/launcher/spark/src/main/java/org/apache/zeppelin/interpreter/launcher/SparkInterpreterLauncher.java
View file

@ -17,6 +17,11 @@
package org.apache.zeppelin.interpreter.launcher;
import java.io.IOException;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.util.stream.StreamSupport;
import org.apache.commons.lang3.StringUtils;
import org.apache.zeppelin.conf.ZeppelinConfiguration;
import org.apache.zeppelin.interpreter.recovery.RecoveryStorage;
@ -76,6 +81,25 @@ public class SparkInterpreterLauncher extends StandardInterpreterLauncher {
!useProxyUserEnv.equals("false"))) {
sparkConfBuilder.append(" --proxy-user " + context.getUserName());
}
Path localRepoPath =
Paths.get(zConf.getInterpreterLocalRepoPath(), context.getInterpreterSettingId());
if (isYarnMode()
&& getDeployMode().equals("cluster")
&& Files.exists(localRepoPath)
&& Files.isDirectory(localRepoPath)) {
try {
StreamSupport.stream(
Files.newDirectoryStream(localRepoPath, entry -> Files.isRegularFile(entry))
.spliterator(),
false)
.map(jar -> jar.toAbsolutePath().toString())
.reduce((x, y) -> x.concat(",").concat(y))
.ifPresent(extraJars -> sparkConfBuilder.append(" --jars ").append(extraJars));
} catch (IOException e) {
LOGGER.error("Cannot make a list of additional jars from localRepo: {}", localRepoPath, e);
}
}
env.put("ZEPPELIN_SPARK_CONF", sparkConfBuilder.toString());

44
zeppelin-plugins/launcher/spark/src/test/java/org/apache/zeppelin/interpreter/launcher/SparkInterpreterLauncherTest.java
View file

@ -17,6 +17,10 @@
package org.apache.zeppelin.interpreter.launcher;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
import org.apache.commons.io.FileUtils;
import org.apache.zeppelin.conf.ZeppelinConfiguration;
import org.apache.zeppelin.interpreter.InterpreterOption;
import org.apache.zeppelin.interpreter.remote.RemoteInterpreterManagedProcess;
@ -177,6 +181,45 @@ public class SparkInterpreterLauncherTest {
InterpreterOption option = new InterpreterOption();
option.setUserImpersonate(true);
InterpreterLaunchContext context = new InterpreterLaunchContext(properties, option, null, "user1", "intpGroupId", "groupId", "spark", "spark", 0, "host");
Path localRepoPath = Paths.get(zConf.getInterpreterLocalRepoPath(), context.getInterpreterSettingId());
FileUtils.deleteDirectory(localRepoPath.toFile());
Files.createDirectories(localRepoPath);
Files.createFile(Paths.get(localRepoPath.toAbsolutePath().toString(), "test.jar"));
InterpreterClient client = launcher.launch(context);
assertTrue( client instanceof RemoteInterpreterManagedProcess);
RemoteInterpreterManagedProcess interpreterProcess = (RemoteInterpreterManagedProcess) client;
assertEquals("spark", interpreterProcess.getInterpreterSettingName());
assertTrue(interpreterProcess.getInterpreterDir().endsWith("/interpreter/spark"));
assertTrue(interpreterProcess.getLocalRepoDir().endsWith("/local-repo/groupId"));
assertEquals(zConf.getInterpreterRemoteRunnerPath(), interpreterProcess.getInterpreterRunner());
assertTrue(interpreterProcess.getEnv().size() >= 3);
assertEquals("/user/spark", interpreterProcess.getEnv().get("SPARK_HOME"));
assertEquals("true", interpreterProcess.getEnv().get("ZEPPELIN_SPARK_YARN_CLUSTER"));
assertEquals(" --master yarn --files .//conf/log4j_yarn_cluster.properties --conf spark.files='file_1' --conf spark.jars='jar_1' --conf spark.submit.deployMode='cluster' --conf spark.yarn.isPython=true --conf spark.yarn.submit.waitAppCompletion=false --proxy-user user1 --jars " + Paths.get(localRepoPath.toAbsolutePath().toString(), "test.jar").toString(), interpreterProcess.getEnv().get("ZEPPELIN_SPARK_CONF"));
Files.deleteIfExists(Paths.get(localRepoPath.toAbsolutePath().toString(), "test.jar"));
FileUtils.deleteDirectory(localRepoPath.toFile());
}
@Test
public void testYarnClusterMode_3() throws IOException {
ZeppelinConfiguration zConf = new ZeppelinConfiguration();
SparkInterpreterLauncher launcher = new SparkInterpreterLauncher(zConf, null);
Properties properties = new Properties();
properties.setProperty("SPARK_HOME", "/user/spark");
properties.setProperty("property_1", "value_1");
properties.setProperty("master", "yarn");
properties.setProperty("spark.submit.deployMode", "cluster");
properties.setProperty("spark.files", "file_1");
properties.setProperty("spark.jars", "jar_1");
InterpreterOption option = new InterpreterOption();
option.setUserImpersonate(true);
InterpreterLaunchContext context = new InterpreterLaunchContext(properties, option, null, "user1", "intpGroupId", "groupId", "spark", "spark", 0, "host");
Path localRepoPath = Paths.get(zConf.getInterpreterLocalRepoPath(), context.getInterpreterSettingId());
FileUtils.deleteDirectory(localRepoPath.toFile());
Files.createDirectories(localRepoPath);
InterpreterClient client = launcher.launch(context);
assertTrue( client instanceof RemoteInterpreterManagedProcess);
RemoteInterpreterManagedProcess interpreterProcess = (RemoteInterpreterManagedProcess) client;
@ -188,5 +231,6 @@ public class SparkInterpreterLauncherTest {
assertEquals("/user/spark", interpreterProcess.getEnv().get("SPARK_HOME"));
assertEquals("true", interpreterProcess.getEnv().get("ZEPPELIN_SPARK_YARN_CLUSTER"));
assertEquals(" --master yarn --files .//conf/log4j_yarn_cluster.properties --conf spark.files='file_1' --conf spark.jars='jar_1' --conf spark.submit.deployMode='cluster' --conf spark.yarn.isPython=true --conf spark.yarn.submit.waitAppCompletion=false --proxy-user user1", interpreterProcess.getEnv().get("ZEPPELIN_SPARK_CONF"));
FileUtils.deleteDirectory(localRepoPath.toFile());
}
}

4
zeppelin-server/src/main/java/org/apache/zeppelin/realm/ActiveDirectoryGroupRealm.java
View file

@ -238,12 +238,14 @@ public class ActiveDirectoryGroupRealm extends AbstractLdapRealm {
return new SimpleAuthorizationInfo(roleNames);
}
public List<String> searchForUserName(String containString, LdapContext ldapContext)
public List<String> searchForUserName(String containString, LdapContext ldapContext,
int numUsersToFetch)
throws NamingException {
List<String> userNameList = new ArrayList<>();
SearchControls searchCtls = new SearchControls();
searchCtls.setSearchScope(SearchControls.SUBTREE_SCOPE);
searchCtls.setCountLimit(numUsersToFetch);
String searchFilter = "(&(objectClass=*)(userPrincipalName=*" + containString + "*))";
Object[] searchArguments = new Object[]{containString};

82
zeppelin-server/src/main/java/org/apache/zeppelin/rest/AdminRestApi.java Normal file
View file

@ -0,0 +1,82 @@
/*
* 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.rest;
import com.google.common.collect.Lists;
import java.util.List;
import javax.ws.rs.BadRequestException;
import javax.ws.rs.GET;
import javax.ws.rs.POST;
import javax.ws.rs.Path;
import javax.ws.rs.QueryParam;
import org.apache.commons.lang.StringUtils;
import org.apache.zeppelin.annotation.ZeppelinApi;
import org.apache.zeppelin.rest.message.LoggerRequest;
import org.apache.zeppelin.service.AdminService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/** This rest apis support some of feature related admin. e.g. changin log level. */
@Path("/admin")
public class AdminRestApi {
private static final Logger logger = LoggerFactory.getLogger(AdminRestApi.class);
private AdminService adminService;
public AdminRestApi(AdminService adminService) {
this.adminService = adminService;
}
/**
* It gets current loggers' name and level.
*
* @param name FQCN
* @return List of current loggers' name and level with json format. It returns all of loggers'
* name and level without name. With name, it returns only specific logger's name and level.
*/
@GET
@ZeppelinApi
public List<org.apache.log4j.Logger> getLoggerSetting(@QueryParam("name") String name) {
logger.debug("name: {}", name);
return null == name || name.isEmpty()
? adminService.getLoggers()
: Lists.newArrayList(adminService.getLogger(name));
}
/**
* It change logger's level.
*
* @param loggerRequest logger's name and level with json format
* @return The changed logger's name and level.
*/
@POST
@ZeppelinApi
public List<org.apache.log4j.Logger> setLoggerLevel(LoggerRequest loggerRequest) {
if (null == loggerRequest
|| StringUtils.isEmpty(loggerRequest.getName())
|| StringUtils.isEmpty(loggerRequest.getLevel())) {
logger.trace("loggerRequest: {}", loggerRequest);
throw new BadRequestException("Wrong request body");
}
logger.debug("loggerRequest: {}", loggerRequest);
adminService.setLoggerLevel(loggerRequest);
return Lists.newArrayList(adminService.getLogger(loggerRequest.getName()));
}
}

45
zeppelin-server/src/main/java/org/apache/zeppelin/rest/CredentialRestApi.java
View file

@ -14,38 +14,31 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.zeppelin.rest;
import com.google.common.base.Strings;
import com.google.gson.Gson;
import com.google.gson.reflect.TypeToken;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.IOException;
import java.util.Map;
import javax.servlet.http.HttpServletRequest;
import javax.ws.rs.DELETE;
import javax.ws.rs.GET;
import javax.ws.rs.PUT;
import javax.ws.rs.Path;
import javax.ws.rs.PathParam;
import javax.ws.rs.Produces;
import javax.ws.rs.core.Context;
import javax.ws.rs.core.Response;
import javax.ws.rs.core.Response.Status;
import org.apache.zeppelin.server.JsonResponse;
import org.apache.zeppelin.user.Credentials;
import org.apache.zeppelin.user.UserCredentials;
import org.apache.zeppelin.user.UsernamePassword;
import org.apache.zeppelin.utils.SecurityUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
* Credential Rest API.
*/
/** Credential Rest API. */
@Path("/credential")
@Produces("application/json")
public class CredentialRestApi {
@ -53,12 +46,6 @@ public class CredentialRestApi {
private Credentials credentials;
private Gson gson = new Gson();
@Context
private HttpServletRequest servReq;
public CredentialRestApi() {
}
public CredentialRestApi(Credentials credentials) {
this.credentials = credentials;
}
@ -73,14 +60,15 @@ public class CredentialRestApi {
*/
@PUT
public Response putCredentials(String message) throws IOException, IllegalArgumentException {
Map<String, String> messageMap = gson.fromJson(message,
new TypeToken<Map<String, String>>(){}.getType());
Map<String, String> messageMap =
gson.fromJson(message, new TypeToken<Map<String, String>>() {}.getType());
String entity = messageMap.get("entity");
String username = messageMap.get("username");
String password = messageMap.get("password");
if (Strings.isNullOrEmpty(entity)
|| Strings.isNullOrEmpty(username) || Strings.isNullOrEmpty(password)) {
|| Strings.isNullOrEmpty(username)
|| Strings.isNullOrEmpty(password)) {
return new JsonResponse(Status.BAD_REQUEST).build();
}
@ -95,31 +83,26 @@ public class CredentialRestApi {
/**
* Get User Credentials list REST API.
*
* @param
* @return JSON with status.OK
* @throws IOException
* @throws IllegalArgumentException
*/
@GET
public Response getCredentials(String message) throws
IOException, IllegalArgumentException {
public Response getCredentials() throws IllegalArgumentException {
String user = SecurityUtils.getPrincipal();
logger.info("getCredentials credentials for user {} ", user);
UserCredentials uc = credentials.getUserCredentials(user);
return new JsonResponse(Status.OK, uc).build();
return new JsonResponse<>(Status.OK, uc).build();
}
/**
* Remove User Credentials REST API.
*
* @param
* @return JSON with status.OK
* @throws IOException
* @throws IllegalArgumentException
*/
@DELETE
public Response removeCredentials(String message) throws
IOException, IllegalArgumentException {
public Response removeCredentials() throws IOException, IllegalArgumentException {
String user = SecurityUtils.getPrincipal();
logger.info("removeCredentials credentials for user {} ", user);
UserCredentials uc = credentials.removeUserCredentials(user);
@ -139,11 +122,11 @@ public class CredentialRestApi {
*/
@DELETE
@Path("{entity}")
public Response removeCredentialEntity(@PathParam("entity") String entity) throws
IOException, IllegalArgumentException {
public Response removeCredentialEntity(@PathParam("entity") String entity)
throws IOException, IllegalArgumentException {
String user = SecurityUtils.getPrincipal();
logger.info("removeCredentialEntity for user {} entity {}", user, entity);
if (credentials.removeCredentialEntity(user, entity) == false) {
if (!credentials.removeCredentialEntity(user, entity)) {
return new JsonResponse(Status.NOT_FOUND).build();
}
return new JsonResponse(Status.OK).build();

13
zeppelin-server/src/main/java/org/apache/zeppelin/rest/GetUserList.java
View file

@ -91,7 +91,7 @@ public class GetUserList {
/**
* Function to extract users from LDAP.
*/
public List<String> getUserList(JndiLdapRealm r, String searchText) {
public List<String> getUserList(JndiLdapRealm r, String searchText, int numUsersToFetch) {
List<String> userList = new ArrayList<>();
String userDnTemplate = r.getUserDnTemplate();
String userDn[] = userDnTemplate.split(",", 2);
@ -101,6 +101,7 @@ public class GetUserList {
try {
LdapContext ctx = cf.getSystemLdapContext();
SearchControls constraints = new SearchControls();
constraints.setCountLimit(numUsersToFetch);
constraints.setSearchScope(SearchControls.SUBTREE_SCOPE);
String[] attrIDs = {userDnPrefix};
constraints.setReturningAttributes(attrIDs);
@ -123,7 +124,7 @@ public class GetUserList {
/**
* Function to extract users from Zeppelin LdapRealm.
*/
public List<String> getUserList(LdapRealm r, String searchText) {
public List<String> getUserList(LdapRealm r, String searchText, int numUsersToFetch) {
List<String> userList = new ArrayList<>();
if (LOG.isDebugEnabled()) {
LOG.debug("SearchText: " + searchText);
@ -136,11 +137,12 @@ public class GetUserList {
LdapContext ctx = cf.getSystemLdapContext();
SearchControls constraints = new SearchControls();
constraints.setSearchScope(SearchControls.SUBTREE_SCOPE);
constraints.setCountLimit(numUsersToFetch);
String[] attrIDs = {userAttribute};
constraints.setReturningAttributes(attrIDs);
NamingEnumeration result = ctx.search(userSearchRealm, "(&(objectclass=" +
userObjectClass + ")("
+ userAttribute + "=" + searchText + "))", constraints);
+ userAttribute + "=*" + searchText + "*))", constraints);
while (result.hasMore()) {
Attributes attrs = ((SearchResult) result.next()).getAttributes();
if (attrs.get(userAttribute) != null) {
@ -187,11 +189,12 @@ public class GetUserList {
return roleList;
}
public List<String> getUserList(ActiveDirectoryGroupRealm r, String searchText) {
public List<String> getUserList(ActiveDirectoryGroupRealm r, String searchText,
int numUsersToFetch) {
List<String> userList = new ArrayList<>();
try {
LdapContext ctx = r.getLdapContextFactory().getSystemLdapContext();
userList = r.searchForUserName(searchText, ctx);
userList = r.searchForUserName(searchText, ctx, numUsersToFetch);
} catch (Exception e) {
LOG.error("Error retrieving User list from ActiveDirectory Realm", e);
}

27
zeppelin-server/src/main/java/org/apache/zeppelin/rest/NotebookRestApi.java
View file

@ -59,6 +59,11 @@ import org.apache.zeppelin.notebook.Paragraph;
import org.apache.zeppelin.rest.exception.BadRequestException;
import org.apache.zeppelin.rest.exception.ForbiddenException;
import org.apache.zeppelin.rest.exception.NotFoundException;
import org.apache.zeppelin.rest.message.CronRequest;
import org.apache.zeppelin.rest.message.NewNoteRequest;
import org.apache.zeppelin.rest.message.NewParagraphRequest;
import org.apache.zeppelin.rest.message.RunParagraphWithParametersRequest;
import org.apache.zeppelin.rest.message.UpdateParagraphRequest;
import org.apache.zeppelin.search.SearchService;
import org.apache.zeppelin.server.JsonResponse;
import org.apache.zeppelin.socket.NotebookServer;
@ -277,7 +282,6 @@ public class NotebookRestApi {
}
/**
*
* Bind a setting to note.
*
* @throws IOException
@ -499,7 +503,7 @@ public class NotebookRestApi {
@Path("{noteId}/paragraph")
@ZeppelinApi
public Response insertParagraph(@PathParam("noteId") String noteId, String message)
throws IOException {
throws IOException {
String user = SecurityUtils.getPrincipal();
LOG.info("insert paragraph {} {}", noteId, message);
@ -550,6 +554,7 @@ public class NotebookRestApi {
*
* @param message json containing the "text" and optionally the "title" of the paragraph, e.g.
* {"text" : "updated text", "title" : "Updated title" }
*
*/
@PUT
@Path("{noteId}/paragraph/{paragraphId}")
@ -699,16 +704,18 @@ public class NotebookRestApi {
@POST
@Path("job/{noteId}")
@ZeppelinApi
public Response runNoteJobs(@PathParam("noteId") String noteId)
throws IOException, IllegalArgumentException {
LOG.info("run note jobs {} ", noteId);
public Response runNoteJobs(@PathParam("noteId") String noteId,
@QueryParam("waitToFinish") Boolean waitToFinish)
throws IOException, IllegalArgumentException {
boolean blocking = waitToFinish == null ? true : waitToFinish.booleanValue();
LOG.info("run note jobs {} waitToFinish: {}", noteId, blocking);
Note note = notebook.getNote(noteId);
AuthenticationInfo subject = new AuthenticationInfo(SecurityUtils.getPrincipal());
checkIfNoteIsNotNull(note);
checkIfUserCanRun(noteId, "Insufficient privileges you cannot run job for this note");
try {
note.runAll(subject, true);
note.runAll(subject, blocking);
} catch (Exception ex) {
LOG.error("Exception from run", ex);
return new JsonResponse<>(Status.PRECONDITION_FAILED,
@ -944,9 +951,9 @@ public class NotebookRestApi {
@Path("cron/{noteId}")
@ZeppelinApi
public Response removeCronJob(@PathParam("noteId") String noteId)
throws IOException, IllegalArgumentException {
throws IOException, IllegalArgumentException {
LOG.info("Remove cron job note {}", noteId);
Note note = notebook.getNote(noteId);
checkIfNoteIsNotNull(note);
checkIfUserIsOwner(noteId,
@ -973,7 +980,7 @@ public class NotebookRestApi {
@Path("cron/{noteId}")
@ZeppelinApi
public Response getCronJob(@PathParam("noteId") String noteId)
throws IOException, IllegalArgumentException {
throws IOException, IllegalArgumentException {
LOG.info("Get cron job note {}", noteId);
Note note = notebook.getNote(noteId);
@ -1010,7 +1017,7 @@ public class NotebookRestApi {
/**
* Get updated note jobs for job manager
* <p>
*
* Return the `Note` change information within the post unix timestamp.
*
* @return JSON with status.OK

12
zeppelin-server/src/main/java/org/apache/zeppelin/rest/SecurityRestApi.java
View file

@ -109,6 +109,8 @@ public class SecurityRestApi {
@GET
@Path("userlist/{searchText}")
public Response getUserList(@PathParam("searchText") final String searchText) {
final int numUsersToFetch = 5;
List<String> usersList = new ArrayList<>();
List<String> rolesList = new ArrayList<>();
try {
@ -125,13 +127,15 @@ public class SecurityRestApi {
usersList.addAll(getUserListObj.getUserList((IniRealm) realm));
rolesList.addAll(getUserListObj.getRolesList((IniRealm) realm));
} else if (name.equals("org.apache.zeppelin.realm.LdapGroupRealm")) {
usersList.addAll(getUserListObj.getUserList((JndiLdapRealm) realm, searchText));
usersList.addAll(getUserListObj.getUserList((JndiLdapRealm) realm, searchText,
numUsersToFetch));
} else if (name.equals("org.apache.zeppelin.realm.LdapRealm")) {
usersList.addAll(getUserListObj.getUserList((LdapRealm) realm, searchText));
usersList.addAll(getUserListObj.getUserList((LdapRealm) realm, searchText,
numUsersToFetch));
rolesList.addAll(getUserListObj.getRolesList((LdapRealm) realm));
} else if (name.equals("org.apache.zeppelin.realm.ActiveDirectoryGroupRealm")) {
usersList.addAll(getUserListObj.getUserList((ActiveDirectoryGroupRealm) realm,
searchText));
searchText, numUsersToFetch));
} else if (name.equals("org.apache.shiro.realm.jdbc.JdbcRealm")) {
usersList.addAll(getUserListObj.getUserList((JdbcRealm) realm));
}
@ -161,7 +165,7 @@ public class SecurityRestApi {
autoSuggestUserList.add(user);
maxLength++;
}
if (maxLength == 5) {
if (maxLength == numUsersToFetch) {
break;
}
}

44
zeppelin-server/src/main/java/org/apache/zeppelin/rest/exception/WebApplicationExceptionMapper.java Normal file
View file

@ -0,0 +1,44 @@
/*
* 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.rest.exception;
import com.google.gson.Gson;
import com.google.gson.GsonBuilder;
import javax.ws.rs.WebApplicationException;
import javax.ws.rs.core.Response;
import javax.ws.rs.ext.ExceptionMapper;
import javax.ws.rs.ext.Provider;
import org.apache.zeppelin.rest.message.gson.ExceptionSerializer;
@Provider
public class WebApplicationExceptionMapper implements ExceptionMapper<WebApplicationException> {
private final Gson gson;
public WebApplicationExceptionMapper() {
GsonBuilder gsonBuilder = new GsonBuilder().enableComplexMapKeySerialization();
gsonBuilder.registerTypeHierarchyAdapter(
Exception.class, new ExceptionSerializer());
this.gson = gsonBuilder.create();
}
@Override
public Response toResponse(WebApplicationException exception) {
return Response.status(exception.getResponse().getStatus())
.entity(gson.toJson(exception)).build();
}
}

41
zeppelin-server/src/main/java/org/apache/zeppelin/rest/message/LoggerRequest.java Normal file
View file

@ -0,0 +1,41 @@
/*
* 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.rest.message;
public class LoggerRequest {
private String name;
private String level;
public LoggerRequest(String name, String level) {
this.name = name;
this.level = level;
}
public String getName() {
return name;
}
public String getLevel() {
return level;
}
@Override
public String toString() {
return "[name: " + name + ", level: " + level + "]";
}
}

42
zeppelin-server/src/main/java/org/apache/zeppelin/rest/message/gson/ExceptionSerializer.java Normal file
View file

@ -0,0 +1,42 @@
/*
* 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.rest.message.gson;
import com.google.gson.JsonElement;
import com.google.gson.JsonObject;
import com.google.gson.JsonSerializationContext;
import com.google.gson.JsonSerializer;
import java.lang.reflect.Type;
import javax.ws.rs.WebApplicationException;
public class ExceptionSerializer implements JsonSerializer<Exception> {
@Override
public JsonElement serialize(
Exception e, Type type, JsonSerializationContext jsonSerializationContext) {
JsonObject jsonObject = new JsonObject();
jsonObject.addProperty("exception", e.getClass().getSimpleName());
jsonObject.addProperty("message", e.getMessage());
if (e instanceof WebApplicationException) {
jsonObject.addProperty("status", ((WebApplicationException) e).getResponse().getStatus());
}
return jsonObject;
}
}

52
zeppelin-server/src/main/java/org/apache/zeppelin/rest/message/gson/LoggerSerializer.java Normal file
View file

@ -0,0 +1,52 @@
/*
* 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.rest.message.gson;
import com.google.gson.JsonElement;
import com.google.gson.JsonObject;
import com.google.gson.JsonSerializationContext;
import com.google.gson.JsonSerializer;
import java.lang.reflect.Type;
import org.apache.commons.lang3.StringUtils;
import org.apache.log4j.Category;
import org.apache.log4j.Level;
import org.apache.log4j.Logger;
public class LoggerSerializer implements JsonSerializer<Logger> {
@Override
public JsonElement serialize(
Logger logger, Type type, JsonSerializationContext jsonSerializationContext) {
JsonObject jsonObject = new JsonObject();
jsonObject.addProperty("name", logger.getName());
jsonObject.addProperty("level", getLoggerLevel(logger));
return jsonObject;
}
private String getLoggerLevel(Category logger) {
if (null == logger) {
return StringUtils.EMPTY;
}
Level level = logger.getLevel();
if (null != level) {
return level.toString();
} else {
return getLoggerLevel(logger.getParent());
}
}
}

7
zeppelin-server/src/main/java/org/apache/zeppelin/server/CorsFilter.java
View file

@ -21,9 +21,6 @@ import org.slf4j.LoggerFactory;
import java.io.IOException;
import java.net.URISyntaxException;
import java.text.DateFormat;
import java.util.Date;
import java.util.Locale;
import javax.servlet.Filter;
import javax.servlet.FilterChain;
@ -75,9 +72,7 @@ public class CorsFilter implements Filter {
response.setHeader("Access-Control-Allow-Credentials", "true");
response.setHeader("Access-Control-Allow-Headers", "authorization,Content-Type");
response.setHeader("Access-Control-Allow-Methods", "POST, GET, OPTIONS, PUT, HEAD, DELETE");
DateFormat fullDateFormatEN =
DateFormat.getDateTimeInstance(DateFormat.FULL, DateFormat.FULL, new Locale("EN", "en"));
response.setHeader("Date", fullDateFormatEN.format(new Date()));
ZeppelinConfiguration zeppelinConfiguration = ZeppelinConfiguration.create();
response.setHeader("X-FRAME-OPTIONS", zeppelinConfiguration.getXFrameOptions());
if (zeppelinConfiguration.useSsl()) {

93
zeppelin-server/src/main/java/org/apache/zeppelin/server/GsonProvider.java Normal file
View file

@ -0,0 +1,93 @@
/*
* 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.server;
import com.google.gson.Gson;
import com.google.gson.GsonBuilder;
import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.io.OutputStream;
import java.io.PrintWriter;
import java.lang.annotation.Annotation;
import java.lang.reflect.Type;
import javax.ws.rs.Consumes;
import javax.ws.rs.Produces;
import javax.ws.rs.WebApplicationException;
import javax.ws.rs.core.MediaType;
import javax.ws.rs.core.MultivaluedMap;
import javax.ws.rs.ext.MessageBodyReader;
import javax.ws.rs.ext.MessageBodyWriter;
import javax.ws.rs.ext.Provider;
import org.apache.log4j.Logger;
import org.apache.zeppelin.rest.message.LoggerRequest;
import org.apache.zeppelin.rest.message.gson.LoggerSerializer;
@Provider
@Produces(MediaType.APPLICATION_JSON)
@Consumes(MediaType.APPLICATION_JSON)
public class GsonProvider<T> implements MessageBodyReader<T>, MessageBodyWriter<T> {
private final Gson gson;
public GsonProvider() {
GsonBuilder gsonBuilder = new GsonBuilder().enableComplexMapKeySerialization();
gsonBuilder.registerTypeAdapter(Logger.class, new LoggerSerializer());
this.gson = gsonBuilder.create();
}
@Override
public boolean isReadable(
Class<?> type, Type genericType, Annotation[] annotations, MediaType mediaType) {
return type == LoggerRequest.class; // For backward compatibility
}
@Override
public T readFrom(
Class<T> type,
Type genericType,
Annotation[] annotations,
MediaType mediaType,
MultivaluedMap<String, String> httpHeaders,
InputStream entityStream)
throws IOException, WebApplicationException {
return gson.fromJson(new BufferedReader(new InputStreamReader(entityStream)), type);
}
@Override
public boolean isWriteable(
Class<?> type, Type genericType, Annotation[] annotations, MediaType mediaType) {
return type != String.class; // Keep backward compatibility
}
@Override
public void writeTo(
T t,
Class<?> type,
Type genericType,
Annotation[] annotations,
MediaType mediaType,
MultivaluedMap<String, Object> httpHeaders,
OutputStream entityStream)
throws IOException, WebApplicationException {
try (PrintWriter printWriter = new PrintWriter(entityStream)) {
printWriter.write(gson.toJson(t));
printWriter.flush();
}
}
}

15
zeppelin-server/src/main/java/org/apache/zeppelin/server/ZeppelinServer.java
View file

@ -31,6 +31,9 @@ import org.apache.shiro.realm.text.IniRealm;
import org.apache.shiro.web.env.EnvironmentLoaderListener;
import org.apache.shiro.web.mgt.DefaultWebSecurityManager;
import org.apache.shiro.web.servlet.ShiroFilter;
import org.apache.zeppelin.rest.AdminRestApi;
import org.apache.zeppelin.rest.exception.WebApplicationExceptionMapper;
import org.apache.zeppelin.service.AdminService;
import org.eclipse.jetty.http.HttpVersion;
import org.eclipse.jetty.server.HttpConfiguration;
import org.eclipse.jetty.server.HttpConnectionFactory;
@ -165,7 +168,7 @@ public class ZeppelinServer extends Application {
notebookWsServer, notebookWsServer);
this.replFactory = new InterpreterFactory(interpreterSettingManager);
this.notebookRepo = new NotebookRepoSync(conf);
this.noteSearchService = new LuceneSearch();
this.noteSearchService = new LuceneSearch(conf);
this.notebookAuthorization = NotebookAuthorization.getInstance();
this.credentials = new Credentials(
conf.credentialsPersist(),
@ -429,6 +432,11 @@ public class ZeppelinServer extends Application {
@Override
public Set<Class<?>> getClasses() {
Set<Class<?>> classes = new HashSet<>();
classes.add(GsonProvider.class);
classes.add(WebApplicationExceptionMapper.class);
return classes;
}
@ -466,6 +474,11 @@ public class ZeppelinServer extends Application {
ConfigurationsRestApi settingsApi = new ConfigurationsRestApi(notebook);
singletons.add(settingsApi);
AdminService adminService = new AdminService();
AdminRestApi adminRestApi = new AdminRestApi(adminService);
singletons.add(adminRestApi);
return singletons;
}

78
zeppelin-server/src/main/java/org/apache/zeppelin/service/AdminService.java Normal file
View file

@ -0,0 +1,78 @@
/*
* 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.service;
import java.util.Enumeration;
import java.util.Iterator;
import java.util.List;
import java.util.Spliterator;
import java.util.Spliterators;
import java.util.stream.Collectors;
import java.util.stream.StreamSupport;
import javax.ws.rs.BadRequestException;
import org.apache.log4j.LogManager;
import org.apache.zeppelin.rest.message.LoggerRequest;
/** This class handles all of business logic of {@link org.apache.zeppelin.rest.AdminRestApi}. */
public class AdminService {
public List<org.apache.log4j.Logger> getLoggers() {
Enumeration loggers = LogManager.getCurrentLoggers();
return StreamSupport.stream(
Spliterators.spliteratorUnknownSize(
new Iterator<org.apache.log4j.Logger>() {
@Override
public boolean hasNext() {
return loggers.hasMoreElements();
}
@Override
public org.apache.log4j.Logger next() {
return org.apache.log4j.Logger.class.cast(loggers.nextElement());
}
},
Spliterator.ORDERED),
false)
.collect(Collectors.toList());
}
public org.apache.log4j.Logger getLogger(String name) {
return LogManager.getLogger(name);
}
public void setLoggerLevel(LoggerRequest loggerRequest) throws BadRequestException {
try {
Class.forName(loggerRequest.getName());
} catch (Throwable ignore) {
throw new BadRequestException(
"The class of '" + loggerRequest.getName() + "' doesn't exists");
}
org.apache.log4j.Logger logger = LogManager.getLogger(loggerRequest.getName());
if (null == logger) {
throw new BadRequestException("The name of the logger is wrong");
}
org.apache.log4j.Level level = org.apache.log4j.Level.toLevel(loggerRequest.getLevel(), null);
if (null == level) {
throw new BadRequestException("The level of the logger is wrong");
}
logger.setLevel(level);
}
}

10
zeppelin-server/src/main/java/org/apache/zeppelin/socket/NotebookServer.java
View file

@ -2518,12 +2518,14 @@ public class NotebookServer extends WebSocketServlet
try {
interpreter = notebook().getInterpreterFactory().getInterpreter(user, noteId, replName);
LOG.debug("getEditorSetting for interpreter: {} for paragraph {}", replName, paragraphId);
resp.put("editor", notebook().getInterpreterSettingManager().
getEditorSetting(interpreter, user, noteId, replName));
conn.send(serializeMessage(resp));
} catch (InterpreterNotFoundException e) {
throw new IOException("Fail to get interpreter: " + replName, e);
LOG.warn("Fail to get interpreter: " + replName);
return;
}
resp.put("editor", notebook().getInterpreterSettingManager().
getEditorSetting(interpreter, user, noteId, replName));
conn.send(serializeMessage(resp));
}
private void getInterpreterSettings(NotebookSocket conn, AuthenticationInfo subject)

127
zeppelin-server/src/test/java/org/apache/zeppelin/rest/CredentialsRestApiTest.java
View file

@ -19,119 +19,80 @@ package org.apache.zeppelin.rest;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertThat;
import com.google.gson.Gson;
import com.google.gson.reflect.TypeToken;
import org.apache.commons.httpclient.methods.DeleteMethod;
import org.apache.commons.httpclient.methods.GetMethod;
import org.apache.commons.httpclient.methods.PutMethod;
import org.junit.AfterClass;
import org.junit.BeforeClass;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.IOException;
import java.nio.file.Files;
import java.util.Map;
import javax.ws.rs.core.Response;
import javax.ws.rs.core.Response.Status;
import org.apache.zeppelin.user.Credentials;
import org.apache.zeppelin.user.UserCredentials;
import org.junit.Before;
import org.junit.Test;
public class CredentialsRestApiTest extends AbstractTestRestApi {
protected static final Logger LOG = LoggerFactory.getLogger(CredentialsRestApiTest.class);
Gson gson = new Gson();
public class CredentialsRestApiTest {
private final Gson gson = new Gson();
@BeforeClass
public static void init() throws Exception {
AbstractTestRestApi.startUp(CredentialsRestApiTest.class.getSimpleName());
}
private CredentialRestApi credentialRestApi;
private Credentials credentials;
@AfterClass
public static void destroy() throws Exception {
AbstractTestRestApi.shutDown();
@Before
public void setUp() throws IOException {
credentials =
new Credentials(false, Files.createTempFile("credentials", "test").toString(), null);
credentialRestApi = new CredentialRestApi(credentials);
}
@Test
public void testInvalidRequest() throws IOException {
String jsonInvalidRequestEntityNull = "{\"entity\" : null, \"username\" : \"test\", " +
"\"password\" : \"testpass\"}";
String jsonInvalidRequestNameNull = "{\"entity\" : \"test\", \"username\" : null, " +
"\"password\" : \"testpass\"}";
String jsonInvalidRequestPasswordNull = "{\"entity\" : \"test\", \"username\" : \"test\", " +
"\"password\" : null}";
String jsonInvalidRequestAllNull = "{\"entity\" : null, \"username\" : null, " +
"\"password\" : null}";
String jsonInvalidRequestEntityNull =
"{\"entity\" : null, \"username\" : \"test\", " + "\"password\" : \"testpass\"}";
String jsonInvalidRequestNameNull =
"{\"entity\" : \"test\", \"username\" : null, " + "\"password\" : \"testpass\"}";
String jsonInvalidRequestPasswordNull =
"{\"entity\" : \"test\", \"username\" : \"test\", " + "\"password\" : null}";
String jsonInvalidRequestAllNull =
"{\"entity\" : null, \"username\" : null, " + "\"password\" : null}";
PutMethod entityNullPut = httpPut("/credential", jsonInvalidRequestEntityNull);
entityNullPut.addRequestHeader("Origin", "http://localhost");
assertThat(entityNullPut, isBadRequest());
entityNullPut.releaseConnection();
Response response = credentialRestApi.putCredentials(jsonInvalidRequestEntityNull);
assertEquals(Status.BAD_REQUEST, response.getStatusInfo().toEnum());
PutMethod nameNullPut = httpPut("/credential", jsonInvalidRequestNameNull);
nameNullPut.addRequestHeader("Origin", "http://localhost");
assertThat(nameNullPut, isBadRequest());
nameNullPut.releaseConnection();
response = credentialRestApi.putCredentials(jsonInvalidRequestNameNull);
assertEquals(Status.BAD_REQUEST, response.getStatusInfo().toEnum());
PutMethod passwordNullPut = httpPut("/credential", jsonInvalidRequestPasswordNull);
passwordNullPut.addRequestHeader("Origin", "http://localhost");
assertThat(passwordNullPut, isBadRequest());
passwordNullPut.releaseConnection();
response = credentialRestApi.putCredentials(jsonInvalidRequestPasswordNull);
assertEquals(Status.BAD_REQUEST, response.getStatusInfo().toEnum());
PutMethod allNullPut = httpPut("/credential", jsonInvalidRequestAllNull);
allNullPut.addRequestHeader("Origin", "http://localhost");
assertThat(allNullPut, isBadRequest());
allNullPut.releaseConnection();
response = credentialRestApi.putCredentials(jsonInvalidRequestAllNull);
assertEquals(Status.BAD_REQUEST, response.getStatusInfo().toEnum());
}
public Map<String, UserCredentials> testGetUserCredentials() throws IOException {
GetMethod getMethod = httpGet("/credential");
getMethod.addRequestHeader("Origin", "http://localhost");
Map<String, Object> resp = gson.fromJson(getMethod.getResponseBodyAsString(),
new TypeToken<Map<String, Object>>(){}.getType());
Response response = credentialRestApi.getCredentials();
Map<String, Object> resp =
gson.fromJson(
response.getEntity().toString(), new TypeToken<Map<String, Object>>() {}.getType());
Map<String, Object> body = (Map<String, Object>) resp.get("body");
Map<String, UserCredentials> credentialMap =
(Map<String, UserCredentials>) body.get("userCredentials");
getMethod.releaseConnection();
(Map<String, UserCredentials>) body.get("userCredentials");
return credentialMap;
}
public void testPutUserCredentials(String requestData) throws IOException {
PutMethod putMethod = httpPut("/credential", requestData);
putMethod.addRequestHeader("Origin", "http://localhost");
assertThat(putMethod, isAllowed());
putMethod.releaseConnection();
}
public void testRemoveUserCredentials() throws IOException {
DeleteMethod deleteMethod = httpDelete("/credential/");
assertThat("Test delete method:", deleteMethod, isAllowed());
deleteMethod.releaseConnection();
}
public void testRemoveCredentialEntity(String entity) throws IOException {
DeleteMethod deleteMethod = httpDelete("/credential/" + entity);
assertThat("Test delete method:", deleteMethod, isAllowed());
deleteMethod.releaseConnection();
}
@Test
public void testCredentialsAPIs() throws IOException {
String requestData1 = "{\"entity\" : \"entityname\", \"username\" : \"myuser\", \"password\" " +
": \"mypass\"}";
String requestData1 =
"{\"entity\" : \"entityname\", \"username\" : \"myuser\", \"password\" " + ": \"mypass\"}";
String entity = "entityname";
Map<String, UserCredentials> credentialMap;
testPutUserCredentials(requestData1);
credentialMap = testGetUserCredentials();
assertNotNull("CredentialMap should be null", credentialMap);
credentialRestApi.putCredentials(requestData1);
assertNotNull("CredentialMap should be null", testGetUserCredentials());
testRemoveCredentialEntity(entity);
credentialMap = testGetUserCredentials();
assertNull("CredentialMap should be null", credentialMap.get("entity1"));
credentialRestApi.removeCredentialEntity(entity);
assertNull("CredentialMap should be null", testGetUserCredentials().get("entity1"));
testRemoveUserCredentials();
credentialMap = testGetUserCredentials();
assertEquals("Compare CredentialMap", credentialMap.toString(), "{}");
credentialRestApi.removeCredentials();
assertEquals("Compare CredentialMap", testGetUserCredentials().toString(), "{}");
}
}

48
zeppelin-server/src/test/java/org/apache/zeppelin/service/AdminServiceTest.java Normal file
View file

@ -0,0 +1,48 @@
/*
* 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.service;
import static org.junit.Assert.assertTrue;
import org.apache.log4j.Level;
import org.junit.Test;
public class AdminServiceTest {
@Test
public void testSetLoggerLevel() {
AdminService adminService = new AdminService();
String testLoggerName = "test";
org.apache.log4j.Logger logger = adminService.getLogger(testLoggerName);
org.apache.log4j.Level level = logger.getLevel();
boolean setInfo = false;
if (org.apache.log4j.Level.INFO == level) {
// if a current level is INFO, set DEBUG to check if it's changed or not
logger.setLevel(org.apache.log4j.Level.DEBUG);
} else {
logger.setLevel(org.apache.log4j.Level.INFO);
setInfo = true;
}
logger = adminService.getLogger(testLoggerName);
assertTrue(
"Level of logger should be changed",
(setInfo && org.apache.log4j.Level.INFO == logger.getLevel())
|| (!setInfo && Level.DEBUG == logger.getLevel()));
}
}

12
zeppelin-web/src/app/notebook/paragraph/result/display-table.css
View file

@ -34,10 +34,20 @@
/** support `height: auto` for cells */
.ui-grid-viewport .ui-grid-cell-contents {
word-wrap: break-word;
white-space: normal !important;
white-space: pre-wrap !important;
border-bottom: 1px solid #e8e8e8;
}
input[class *=coltuiGrid] {
white-space: pre-wrap !important;
}
.ui-grid-zeppelin-special-textarea {
width: 100%;
overflow: hidden;
resize: none;
}
.ui-grid-row, .ui-grid-cell {
height: auto !important;
}

31
zeppelin-web/src/app/visualization/builtins/visualization-table.js
View file

@ -116,14 +116,24 @@ export default class TableVisualization extends Visualization {
type: DefaultTableColumnType,
cellTemplate: `
<div ng-if="!grid.getCellValue(row, col).startsWith('%html')"
class="ui-grid-cell-contents">
{{grid.getCellValue(row, col)}}
</div>
class="ui-grid-cell-contents"><span>{{grid.getCellValue(row, col)}}</span></div>
<div ng-if="grid.getCellValue(row, col).startsWith('%html')"
ng-bind-html="grid.getCellValue(row, col).split('%html')[1]"
class="ui-grid-cell-contents">
</div>
`,
</div>`,
editableCellTemplate:
`<div>
<form
name="inputForm">
<textarea
class="ui-grid-zeppelin-special-textarea"
type="INPUT_TYPE"
ng-class="'colt' + col.uid"
ui-grid-editor
ng-model="MODEL_COL_FIELD" />
</form>
</div>
`,
minWidth: this.getColumnMinWidth(colName),
width: '*',
sortingAlgorithm: function(a, b, row1, row2, sortType, gridCol) {
@ -344,6 +354,17 @@ export default class TableVisualization extends Visualization {
gridApi.colResizable.on.columnSizeChanged(scope, () => {
self.persistConfigWithGridState(self.config);
});
gridApi.edit.on.beginCellEdit(scope, function(rowEntity, colDef, triggerEvent) {
let textArea = triggerEvent.currentTarget.children[1].children[0].children[0];
textArea.style.height = textArea.scrollHeight + 'px';
textArea.addEventListener('keydown', function() {
let elem = this;
setTimeout(function() {
elem.style.height = 'auto';
elem.style.height = elem.scrollHeight + 'px';
});
}, 0);
});
// pagination doesn't follow usual life-cycle in ui-grid v4.0.4
// gridApi.pagination.on.paginationChanged(scope, () => { self.persistConfigWithGridState(self.config) })

7
zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreter.java
View file

@ -395,9 +395,10 @@ public class RemoteInterpreter extends Interpreter {
}
private RemoteInterpreterContext convert(InterpreterContext ic) {
return new RemoteInterpreterContext(ic.getNoteId(), ic.getParagraphId(), ic.getReplName(),
ic.getParagraphTitle(), ic.getParagraphText(), gson.toJson(ic.getAuthenticationInfo()),
gson.toJson(ic.getConfig()), ic.getGui().toJson(), gson.toJson(ic.getNoteGui()),
return new RemoteInterpreterContext(ic.getNoteId(), ic.getNoteName(), ic.getParagraphId(),
ic.getReplName(), ic.getParagraphTitle(), ic.getParagraphText(),
gson.toJson(ic.getAuthenticationInfo()), gson.toJson(ic.getConfig()), ic.getGui().toJson(),
gson.toJson(ic.getNoteGui()),
ic.getLocalProperties());
}

4
zeppelin-zengine/src/main/java/org/apache/zeppelin/notebook/Paragraph.java
View file

@ -596,6 +596,7 @@ public class Paragraph extends Job implements Cloneable, JsonSerializable {
InterpreterContext interpreterContext =
InterpreterContext.builder()
.setNoteId(note.getId())
.setNoteName(note.getName())
.setParagraphId(getId())
.setReplName(intpText)
.setParagraphTitle(title)
@ -639,6 +640,7 @@ public class Paragraph extends Job implements Cloneable, JsonSerializable {
InterpreterContext interpreterContext =
InterpreterContext.builder()
.setNoteId(note.getId())
.setNoteName(note.getName())
.setParagraphId(getId())
.setReplName(intpText)
.setParagraphTitle(title)
@ -670,7 +672,7 @@ public class Paragraph extends Job implements Cloneable, JsonSerializable {
private transient Note note;
public ParagraphRunner(Note note, String noteId, String paragraphId) {
ParagraphRunner(Note note, String noteId, String paragraphId) {
super(noteId, paragraphId);
this.note = note;
}

183
zeppelin-zengine/src/main/java/org/apache/zeppelin/search/LuceneSearch.java
View file

@ -16,14 +16,20 @@
*/
package org.apache.zeppelin.search;
import com.google.common.base.Joiner;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.Lists;
import java.io.IOException;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.util.Collection;
import java.util.Collections;
import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;
import org.apache.commons.io.FileUtils;
import org.apache.lucene.analysis.Analyzer;
import org.apache.lucene.analysis.TokenStream;
import org.apache.lucene.analysis.standard.StandardAnalyzer;
@ -39,7 +45,6 @@ import org.apache.lucene.index.IndexWriterConfig;
import org.apache.lucene.index.Term;
import org.apache.lucene.queryparser.classic.MultiFieldQueryParser;
import org.apache.lucene.queryparser.classic.ParseException;
import org.apache.lucene.queryparser.classic.QueryParser;
import org.apache.lucene.search.IndexSearcher;
import org.apache.lucene.search.Query;
import org.apache.lucene.search.ScoreDoc;
@ -51,43 +56,50 @@ import org.apache.lucene.search.highlight.SimpleHTMLFormatter;
import org.apache.lucene.search.highlight.TextFragment;
import org.apache.lucene.search.highlight.TokenSources;
import org.apache.lucene.store.Directory;
import org.apache.lucene.store.MMapDirectory;
import org.apache.lucene.store.RAMDirectory;
import org.apache.zeppelin.conf.ZeppelinConfiguration;
import org.apache.zeppelin.notebook.Note;
import org.apache.zeppelin.notebook.Paragraph;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.google.common.base.Joiner;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.Lists;
/**
* Search (both, indexing and query) the notebooks using Lucene.
*
* Query is thread-safe, as creates new IndexReader every time.
* Index is thread-safe, as re-uses single IndexWriter, which is thread-safe.
* Search (both, indexing and query) the notebooks using Lucene. Query is thread-safe, as creates
* new IndexReader every time. Index is thread-safe, as re-uses single IndexWriter, which is
* thread-safe.
*/
public class LuceneSearch implements SearchService {
private static final Logger LOG = LoggerFactory.getLogger(LuceneSearch.class);
private static final Logger logger = LoggerFactory.getLogger(LuceneSearch.class);
private static final String SEARCH_FIELD_TEXT = "contents";
private static final String SEARCH_FIELD_TITLE = "header";
static final String PARAGRAPH = "paragraph";
static final String ID_FIELD = "id";
private static final String PARAGRAPH = "paragraph";
private static final String ID_FIELD = "id";
Directory ramDirectory;
Analyzer analyzer;
IndexWriterConfig iwc;
IndexWriter writer;
private Directory directory;
private Path directoryPath;
private Analyzer analyzer;
private IndexWriterConfig indexWriterConfig;
private IndexWriter indexWriter;
public LuceneSearch() {
ramDirectory = new RAMDirectory();
analyzer = new StandardAnalyzer();
iwc = new IndexWriterConfig(analyzer);
public LuceneSearch(ZeppelinConfiguration zeppelinConfiguration) {
try {
writer = new IndexWriter(ramDirectory, iwc);
this.directoryPath =
Files.createTempDirectory(
Paths.get(zeppelinConfiguration.getZeppelinSearchTempPath()), "zeppelin-search-");
this.directory = new MMapDirectory(directoryPath);
} catch (IOException e) {
LOG.error("Failed to create new IndexWriter", e);
logger.error(
"Failed to create temporary directory for search service. Use memory instead", e);
this.directory = new RAMDirectory();
}
this.analyzer = new StandardAnalyzer();
this.indexWriterConfig = new IndexWriterConfig(analyzer);
try {
this.indexWriter = new IndexWriter(directory, indexWriterConfig);
} catch (IOException e) {
logger.error("Failed to create new IndexWriter", e);
}
}
@ -96,51 +108,49 @@ public class LuceneSearch implements SearchService {
*/
@Override
public List<Map<String, String>> query(String queryStr) {
if (null == ramDirectory) {
if (null == directory) {
throw new IllegalStateException(
"Something went wrong on instance creation time, index dir is null");
}
List<Map<String, String>> result = Collections.emptyList();
try (IndexReader indexReader = DirectoryReader.open(ramDirectory)) {
try (IndexReader indexReader = DirectoryReader.open(directory)) {
IndexSearcher indexSearcher = new IndexSearcher(indexReader);
Analyzer analyzer = new StandardAnalyzer();
MultiFieldQueryParser parser = new MultiFieldQueryParser(
new String[] {SEARCH_FIELD_TEXT, SEARCH_FIELD_TITLE},
analyzer);
MultiFieldQueryParser parser =
new MultiFieldQueryParser(new String[] {SEARCH_FIELD_TEXT, SEARCH_FIELD_TITLE}, analyzer);
Query query = parser.parse(queryStr);
LOG.debug("Searching for: " + query.toString(SEARCH_FIELD_TEXT));
logger.debug("Searching for: " + query.toString(SEARCH_FIELD_TEXT));
SimpleHTMLFormatter htmlFormatter = new SimpleHTMLFormatter();
Highlighter highlighter = new Highlighter(htmlFormatter, new QueryScorer(query));
result = doSearch(indexSearcher, query, analyzer, highlighter);
indexReader.close();
} catch (IOException e) {
LOG.error("Failed to open index dir {}, make sure indexing finished OK", ramDirectory, e);
logger.error("Failed to open index dir {}, make sure indexing finished OK", directory, e);
} catch (ParseException e) {
LOG.error("Failed to parse query " + queryStr, e);
logger.error("Failed to parse query " + queryStr, e);
}
return result;
}
private List<Map<String, String>> doSearch(IndexSearcher searcher, Query query,
Analyzer analyzer, Highlighter highlighter) {
private List<Map<String, String>> doSearch(
IndexSearcher searcher, Query query, Analyzer analyzer, Highlighter highlighter) {
List<Map<String, String>> matchingParagraphs = Lists.newArrayList();
ScoreDoc[] hits;
try {
hits = searcher.search(query, 20).scoreDocs;
for (int i = 0; i < hits.length; i++) {
LOG.debug("doc={} score={}", hits[i].doc, hits[i].score);
logger.debug("doc={} score={}", hits[i].doc, hits[i].score);
int id = hits[i].doc;
Document doc = searcher.doc(id);
String path = doc.get(ID_FIELD);
if (path != null) {
LOG.debug((i + 1) + ". " + path);
logger.debug((i + 1) + ". " + path);
String title = doc.get("title");
if (title != null) {
LOG.debug(" Title: {}", doc.get("title"));
logger.debug(" Title: {}", doc.get("title"));
}
String text = doc.get(SEARCH_FIELD_TEXT);
@ -148,34 +158,38 @@ public class LuceneSearch implements SearchService {
String fragment = "";
if (text != null) {
TokenStream tokenStream = TokenSources.getTokenStream(searcher.getIndexReader(), id,
SEARCH_FIELD_TEXT, analyzer);
TokenStream tokenStream =
TokenSources.getTokenStream(
searcher.getIndexReader(), id, SEARCH_FIELD_TEXT, analyzer);
TextFragment[] frag = highlighter.getBestTextFragments(tokenStream, text, true, 3);
LOG.debug(" {} fragments found for query '{}'", frag.length, query);
logger.debug(" {} fragments found for query '{}'", frag.length, query);
for (int j = 0; j < frag.length; j++) {
if ((frag[j] != null) && (frag[j].getScore() > 0)) {
LOG.debug(" Fragment: {}", frag[j].toString());
logger.debug(" Fragment: {}", frag[j].toString());
}
}
fragment = (frag != null && frag.length > 0) ? frag[0].toString() : "";
}
if (header != null) {
TokenStream tokenTitle = TokenSources.getTokenStream(searcher.getIndexReader(), id,
SEARCH_FIELD_TITLE, analyzer);
TokenStream tokenTitle =
TokenSources.getTokenStream(
searcher.getIndexReader(), id, SEARCH_FIELD_TITLE, analyzer);
TextFragment[] frgTitle = highlighter.getBestTextFragments(tokenTitle, header, true, 3);
header = (frgTitle != null && frgTitle.length > 0) ? frgTitle[0].toString() : "";
} else {
header = "";
}
matchingParagraphs.add(ImmutableMap.of("id", path, // <noteId>/paragraph/<paragraphId>
"name", title, "snippet", fragment, "text", text, "header", header));
matchingParagraphs.add(
ImmutableMap.of(
"id", path, // <noteId>/paragraph/<paragraphId>
"name", title, "snippet", fragment, "text", text, "header", header));
} else {
LOG.info("{}. No {} for this document", i + 1, ID_FIELD);
logger.info("{}. No {} for this document", i + 1, ID_FIELD);
}
}
} catch (IOException | InvalidTokenOffsetsException e) {
LOG.error("Exception on searching for {}", query, e);
logger.error("Exception on searching for {}", query, e);
}
return matchingParagraphs;
}
@ -186,7 +200,7 @@ public class LuceneSearch implements SearchService {
@Override
public void updateIndexDoc(Note note) throws IOException {
updateIndexNoteName(note);
for (Paragraph p: note.getParagraphs()) {
for (Paragraph p : note.getParagraphs()) {
updateIndexParagraph(note, p);
}
}
@ -194,9 +208,9 @@ public class LuceneSearch implements SearchService {
private void updateIndexNoteName(Note note) throws IOException {
String noteName = note.getName();
String noteId = note.getId();
LOG.debug("Indexing Notebook {}, '{}'", noteId, noteName);
logger.debug("Indexing Notebook {}, '{}'", noteId, noteName);
if (null == noteName || noteName.isEmpty()) {
LOG.debug("Skipping empty notebook name");
logger.debug("Skipping empty notebook name");
return;
}
updateDoc(noteId, noteName, null);
@ -204,15 +218,15 @@ public class LuceneSearch implements SearchService {
private void updateIndexParagraph(Note note, Paragraph p) throws IOException {
if (p.getText() == null) {
LOG.debug("Skipping empty paragraph");
logger.debug("Skipping empty paragraph");
return;
}
updateDoc(note.getId(), note.getName(), p);
}
/**
* Updates index for the given note: either note.name or a paragraph If
* paragraph is <code>null</code> - updates only for the note.name
* Updates index for the given note: either note.name or a paragraph If paragraph is <code>null
* </code> - updates only for the note.name
*
* @param noteId
* @param noteName
@ -223,16 +237,16 @@ public class LuceneSearch implements SearchService {
String id = formatId(noteId, p);
Document doc = newDocument(id, noteName, p);
try {
writer.updateDocument(new Term(ID_FIELD, id), doc);
writer.commit();
indexWriter.updateDocument(new Term(ID_FIELD, id), doc);
indexWriter.commit();
} catch (IOException e) {
LOG.error("Failed to updaet index of notebook {}", noteId, e);
logger.error("Failed to updaet index of notebook {}", noteId, e);
}
}
/**
* If paragraph is not null, id is <noteId>/paragraphs/<paragraphId>,
* otherwise it's just <noteId>.
* If paragraph is not null, id is <noteId>/paragraphs/<paragraphId>, otherwise it's just
* <noteId>.
*/
static String formatId(String noteId, Paragraph p) {
String id = noteId;
@ -253,8 +267,7 @@ public class LuceneSearch implements SearchService {
}
/**
* If paragraph is not null, indexes code in the paragraph, otherwise indexes
* the notebook name.
* If paragraph is not null, indexes code in the paragraph, otherwise indexes the notebook name.
*
* @param id id of the document, different for Note name and paragraph
* @param noteName name of the note
@ -294,15 +307,17 @@ public class LuceneSearch implements SearchService {
docsIndexed++;
}
} catch (IOException e) {
LOG.error("Failed to index all Notebooks", e);
logger.error("Failed to index all Notebooks", e);
} finally {
try { // save what's been indexed, even if not full collection
writer.commit();
indexWriter.commit();
} catch (IOException e) {
LOG.error("Failed to save index", e);
logger.error("Failed to save index", e);
}
long end = System.nanoTime();
LOG.info("Indexing {} notebooks took {}ms", docsIndexed,
logger.info(
"Indexing {} notebooks took {}ms",
docsIndexed,
TimeUnit.NANOSECONDS.toMillis(end - start));
}
}
@ -314,9 +329,9 @@ public class LuceneSearch implements SearchService {
public void addIndexDoc(Note note) {
try {
addIndexDocAsync(note);
writer.commit();
indexWriter.commit();
} catch (IOException e) {
LOG.error("Failed to add note {} to index", note, e);
logger.error("Failed to add note {} to index", note, e);
}
}
@ -327,13 +342,13 @@ public class LuceneSearch implements SearchService {
* @throws IOException
*/
private void addIndexDocAsync(Note note) throws IOException {
indexNoteName(writer, note.getId(), note.getName());
indexNoteName(indexWriter, note.getId(), note.getName());
for (Paragraph doc : note.getParagraphs()) {
if (doc.getText() == null) {
LOG.debug("Skipping empty paragraph");
logger.debug("Skipping empty paragraph");
continue;
}
indexDoc(writer, note.getId(), note.getName(), doc);
indexDoc(indexWriter, note.getId(), note.getName(), doc);
}
}
@ -356,18 +371,18 @@ public class LuceneSearch implements SearchService {
private void deleteDoc(Note note, Paragraph p) {
if (null == note) {
LOG.error("Trying to delete note by reference to NULL");
logger.error("Trying to delete note by reference to NULL");
return;
}
String fullNoteOrJustParagraph = formatDeleteId(note.getId(), p);
LOG.debug("Deleting note {}, out of: {}", note.getId(), writer.numDocs());
logger.debug("Deleting note {}, out of: {}", note.getId(), indexWriter.numDocs());
try {
writer.deleteDocuments(new WildcardQuery(new Term(ID_FIELD, fullNoteOrJustParagraph)));
writer.commit();
indexWriter.deleteDocuments(new WildcardQuery(new Term(ID_FIELD, fullNoteOrJustParagraph)));
indexWriter.commit();
} catch (IOException e) {
LOG.error("Failed to delete {} from index by '{}'", note, fullNoteOrJustParagraph, e);
logger.error("Failed to delete {} from index by '{}'", note, fullNoteOrJustParagraph, e);
}
LOG.debug("Done, index contains {} docs now" + writer.numDocs());
logger.debug("Done, index contains {} docs now" + indexWriter.numDocs());
}
/* (non-Javadoc)
@ -376,9 +391,12 @@ public class LuceneSearch implements SearchService {
@Override
public void close() {
try {
writer.close();
indexWriter.close();
if (null != directoryPath) {
FileUtils.deleteDirectory(directoryPath.toFile());
}
} catch (IOException e) {
LOG.error("Failed to .close() the notebook index", e);
logger.error("Failed to .close() the notebook index", e);
}
}
@ -388,24 +406,19 @@ public class LuceneSearch implements SearchService {
* @throws IOException
*/
private void indexNoteName(IndexWriter w, String noteId, String noteName) throws IOException {
LOG.debug("Indexing Notebook {}, '{}'", noteId, noteName);
logger.debug("Indexing Notebook {}, '{}'", noteId, noteName);
if (null == noteName || noteName.isEmpty()) {
LOG.debug("Skipping empty notebook name");
logger.debug("Skipping empty notebook name");
return;
}
indexDoc(w, noteId, noteName, null);
}
/**
* Indexes a single document:
* - code of the paragraph (if non-null)
* - or just a note name
*/
/** Indexes a single document: - code of the paragraph (if non-null) - or just a note name */
private void indexDoc(IndexWriter w, String noteId, String noteName, Paragraph p)
throws IOException {
String id = formatId(noteId, p);
Document doc = newDocument(id, noteName, p);
w.addDocument(doc);
}
}

40
zeppelin-zengine/src/main/java/org/apache/zeppelin/storage/LocalConfigStorage.java
View file

@ -28,7 +28,11 @@ import java.io.File;
import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.io.IOException;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.FileSystems;
import java.nio.file.FileSystem;
import java.nio.file.StandardCopyOption;
/**
* Storing config in local file system
@ -51,7 +55,7 @@ public class LocalConfigStorage extends ConfigStorage {
@Override
public void save(InterpreterInfoSaving settingInfos) throws IOException {
LOGGER.info("Save Interpreter Setting to " + interpreterSettingPath.getAbsolutePath());
writeToFile(settingInfos.toJson(), interpreterSettingPath);
atomicWriteToFile(settingInfos.toJson(), interpreterSettingPath);
}
@Override
@ -68,7 +72,7 @@ public class LocalConfigStorage extends ConfigStorage {
@Override
public void save(NotebookAuthorizationInfoSaving authorizationInfoSaving) throws IOException {
LOGGER.info("Save notebook authorization to file: " + authorizationPath);
writeToFile(authorizationInfoSaving.toJson(), authorizationPath);
atomicWriteToFile(authorizationInfoSaving.toJson(), authorizationPath);
}
@Override
@ -95,17 +99,37 @@ public class LocalConfigStorage extends ConfigStorage {
@Override
public void saveCredentials(String credentials) throws IOException {
LOGGER.info("Save Credentials to file: " + credentialPath);
writeToFile(credentials, credentialPath);
atomicWriteToFile(credentials, credentialPath);
}
private String readFromFile(File file) throws IOException {
return IOUtils.toString(new FileInputStream(file));
}
private void writeToFile(String content, File file) throws IOException {
FileOutputStream out = new FileOutputStream(file);
IOUtils.write(content, out);
private void atomicWriteToFile(String content, File file) throws IOException {
File directory = file.getParentFile();
File tempFile = File.createTempFile(file.getName(), null, directory);
FileOutputStream out = new FileOutputStream(tempFile);
try {
IOUtils.write(content, out);
} catch (IOException iox) {
if (!tempFile.delete()) {
tempFile.deleteOnExit();
}
throw iox;
}
out.close();
FileSystem defaultFileSystem = FileSystems.getDefault();
Path tempFilePath = defaultFileSystem.getPath(tempFile.getCanonicalPath());
Path destinationFilePath = defaultFileSystem.getPath(file.getCanonicalPath());
try {
Files.move(tempFilePath, destinationFilePath, StandardCopyOption.ATOMIC_MOVE);
} catch (IOException iox) {
if (!tempFile.delete()) {
tempFile.deleteOnExit();
}
throw iox;
}
}
}
}

34
zeppelin-zengine/src/test/java/org/apache/zeppelin/interpreter/SparkDownloadUtils.java
View file

@ -1,17 +1,16 @@
package org.apache.zeppelin.interpreter;
import org.apache.commons.io.FileUtils;
import org.apache.commons.io.IOUtils;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.BufferedReader;
import java.io.File;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.net.URL;
import org.apache.commons.io.FileUtils;
import org.apache.commons.io.IOUtils;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
public class SparkDownloadUtils {
private static Logger LOGGER = LoggerFactory.getLogger(SparkDownloadUtils.class);
@ -34,18 +33,39 @@ public class SparkDownloadUtils {
return targetSparkHomeFolder.getAbsolutePath();
}
// Try mirrors a few times until one succeeds
boolean downloaded = false;
for (int i = 0; i < 3; i++) {
try {
String preferredMirror = IOUtils.toString(new URL("https://www.apache.org/dyn/closer.lua?preferred=true"));
File downloadFile = new File(downloadFolder + "/spark-" + version + "-bin-hadoop2.6.tgz");
String downloadURL = preferredMirror + "/spark/spark-" + version + "/spark-" + version + "-bin-hadoop2.6.tgz";;
String downloadURL = preferredMirror + "/spark/spark-" + version + "/spark-" + version + "-bin-hadoop2.6.tgz";
runShellCommand(new String[] {"wget", downloadURL, "-P", downloadFolder});
runShellCommand(new String[]{"tar", "-xvf", downloadFile.getAbsolutePath(), "-C", downloadFolder});
downloaded = true;
break;
} catch (Exception e) {
LOGGER.warn("Failed to download Spark", e);
}
}
// fallback to use apache archive
// https://archive.apache.org/dist/spark/spark-1.6.3/spark-1.6.3-bin-hadoop2.6.tgz
if (!downloaded) {
File downloadFile = new File(downloadFolder + "/spark-" + version + "-bin-hadoop2.6.tgz");
String downloadURL =
"https://archive.apache.org/dist/spark/spark-"
+ version
+ "/spark-"
+ version
+ "-bin-hadoop2.6.tgz";
try {
runShellCommand(new String[] {"wget", downloadURL, "-P", downloadFolder});
runShellCommand(
new String[] {"tar", "-xvf", downloadFile.getAbsolutePath(), "-C", downloadFolder});
} catch (Exception e) {
throw new RuntimeException("Fail to download spark " + version, e);
}
}
return targetSparkHomeFolder.getAbsolutePath();
}

147
zeppelin-zengine/src/test/java/org/apache/zeppelin/search/LuceneSearchTest.java
View file

@ -17,14 +17,15 @@
package org.apache.zeppelin.search;
import static com.google.common.truth.Truth.assertThat;
import static org.mockito.Mockito.*;
import static org.apache.zeppelin.search.LuceneSearch.formatId;
import static org.mockito.Mockito.mock;
import com.google.common.base.Splitter;
import java.io.IOException;
import java.util.Arrays;
import java.util.List;
import java.util.Map;
import org.apache.zeppelin.conf.ZeppelinConfiguration;
import org.apache.zeppelin.interpreter.InterpreterFactory;
import org.apache.zeppelin.interpreter.InterpreterSettingManager;
import org.apache.zeppelin.notebook.Note;
@ -36,14 +37,12 @@ import org.junit.Before;
import org.junit.BeforeClass;
import org.junit.Test;
import com.google.common.base.Splitter;
public class LuceneSearchTest {
private static NotebookRepo notebookRepoMock;
private static InterpreterFactory interpreterFactory;
private static InterpreterSettingManager interpreterSettingManager;
private SearchService noteSearchService;
private AuthenticationInfo anonymous;
@ -53,13 +52,13 @@ public class LuceneSearchTest {
interpreterFactory = mock(InterpreterFactory.class);
interpreterSettingManager = mock(InterpreterSettingManager.class);
// when(replLoaderMock.getInterpreterSettings())
// .thenReturn(ImmutableList.<InterpreterSetting>of());
// when(replLoaderMock.getInterpreterSettings())
// .thenReturn(ImmutableList.<InterpreterSetting>of());
}
@Before
public void startUp() {
noteSearchService = new LuceneSearch();
noteSearchService = new LuceneSearch(ZeppelinConfiguration.create());
anonymous = new AuthenticationInfo("anonymous");
}
@ -68,42 +67,45 @@ public class LuceneSearchTest {
noteSearchService.close();
}
@Test public void canIndexNotebook() {
//give
@Test
public void canIndexNotebook() {
// give
Note note1 = newNoteWithParagraph("Notebook1", "test");
Note note2 = newNoteWithParagraph("Notebook2", "not test");
List<Note> notebook = Arrays.asList(note1, note2);
//when
// when
noteSearchService.addIndexDocs(notebook);
}
@Test public void canIndexAndQuery() {
//given
@Test
public void canIndexAndQuery() {
// given
Note note1 = newNoteWithParagraph("Notebook1", "test");
Note note2 = newNoteWithParagraphs("Notebook2", "not test", "not test at all");
noteSearchService.addIndexDocs(Arrays.asList(note1, note2));
//when
// when
List<Map<String, String>> results = noteSearchService.query("all");
//then
// then
assertThat(results).isNotEmpty();
assertThat(results.size()).isEqualTo(1);
assertThat(results.get(0))
.containsEntry("id", formatId(note2.getId(), note2.getLastParagraph()));
.containsEntry("id", formatId(note2.getId(), note2.getLastParagraph()));
}
@Test public void canIndexAndQueryByNotebookName() {
//given
@Test
public void canIndexAndQueryByNotebookName() {
// given
Note note1 = newNoteWithParagraph("Notebook1", "test");
Note note2 = newNoteWithParagraphs("Notebook2", "not test", "not test at all");
noteSearchService.addIndexDocs(Arrays.asList(note1, note2));
//when
// when
List<Map<String, String>> results = noteSearchService.query("Notebook1");
//then
// then
assertThat(results).isNotEmpty();
assertThat(results.size()).isEqualTo(1);
assertThat(results.get(0)).containsEntry("id", note1.getId());
@ -111,15 +113,15 @@ public class LuceneSearchTest {
@Test
public void canIndexAndQueryByParagraphTitle() {
//given
// given
Note note1 = newNoteWithParagraph("Notebook1", "test", "testingTitleSearch");
Note note2 = newNoteWithParagraph("Notebook2", "not test", "notTestingTitleSearch");
noteSearchService.addIndexDocs(Arrays.asList(note1, note2));
//when
// when
List<Map<String, String>> results = noteSearchService.query("testingTitleSearch");
//then
// then
assertThat(results).isNotEmpty();
assertThat(results.size()).isAtLeast(1);
int TitleHits = 0;
@ -131,41 +133,44 @@ public class LuceneSearchTest {
assertThat(TitleHits).isAtLeast(1);
}
@Test public void indexKeyContract() throws IOException {
//give
@Test
public void indexKeyContract() throws IOException {
// give
Note note1 = newNoteWithParagraph("Notebook1", "test");
//when
// when
noteSearchService.addIndexDoc(note1);
//then
String id = resultForQuery("test").get(0).get(LuceneSearch.ID_FIELD);
// then
String id = resultForQuery("test").get(0).get("id"); // LuceneSearch.ID_FIELD
assertThat(Splitter.on("/").split(id)) //key structure <noteId>/paragraph/<paragraphId>
.containsAllOf(note1.getId(), LuceneSearch.PARAGRAPH, note1.getLastParagraph().getId());
assertThat(Splitter.on("/").split(id)) // key structure <noteId>/paragraph/<paragraphId>
.containsAllOf(
note1.getId(), "paragraph", note1.getLastParagraph().getId()); // LuceneSearch.PARAGRAPH
}
@Test //(expected=IllegalStateException.class)
@Test // (expected=IllegalStateException.class)
public void canNotSearchBeforeIndexing() {
//given NO noteSearchService.index() was called
//when
// given NO noteSearchService.index() was called
// when
List<Map<String, String>> result = noteSearchService.query("anything");
//then
// then
assertThat(result).isEmpty();
//assert logs were printed
//"ERROR org.apache.zeppelin.search.SearchService:97 - Failed to open index dir RAMDirectory"
// assert logs were printed
// "ERROR org.apache.zeppelin.search.SearchService:97 - Failed to open index dir RAMDirectory"
}
@Test public void canIndexAndReIndex() throws IOException {
//given
@Test
public void canIndexAndReIndex() throws IOException {
// given
Note note1 = newNoteWithParagraph("Notebook1", "test");
Note note2 = newNoteWithParagraphs("Notebook2", "not test", "not test at all");
noteSearchService.addIndexDocs(Arrays.asList(note1, note2));
//when
// when
Paragraph p2 = note2.getLastParagraph();
p2.setText("test indeed");
noteSearchService.updateIndexDoc(note2);
//then
// then
List<Map<String, String>> results = noteSearchService.query("all");
assertThat(results).isEmpty();
@ -173,24 +178,26 @@ public class LuceneSearchTest {
assertThat(results).isNotEmpty();
}
@Test public void canDeleteNull() throws IOException {
//give
@Test
public void canDeleteNull() throws IOException {
// give
// looks like a bug in web UI: it tries to delete a note twice (after it has just been deleted)
//when
// when
noteSearchService.deleteIndexDocs(null);
}
@Test public void canDeleteFromIndex() throws IOException {
//given
@Test
public void canDeleteFromIndex() throws IOException {
// given
Note note1 = newNoteWithParagraph("Notebook1", "test");
Note note2 = newNoteWithParagraphs("Notebook2", "not test", "not test at all");
noteSearchService.addIndexDocs(Arrays.asList(note1, note2));
assertThat(resultForQuery("Notebook2")).isNotEmpty();
//when
// when
noteSearchService.deleteIndexDocs(note2);
//then
// then
assertThat(noteSearchService.query("all")).isEmpty();
assertThat(resultForQuery("Notebook2")).isEmpty();
@ -199,43 +206,46 @@ public class LuceneSearchTest {
assertThat(results.size()).isEqualTo(1);
}
@Test public void indexParagraphUpdatedOnNoteSave() throws IOException {
//given: total 2 notebooks, 3 paragraphs
@Test
public void indexParagraphUpdatedOnNoteSave() throws IOException {
// given: total 2 notebooks, 3 paragraphs
Note note1 = newNoteWithParagraph("Notebook1", "test");
Note note2 = newNoteWithParagraphs("Notebook2", "not test", "not test at all");
noteSearchService.addIndexDocs(Arrays.asList(note1, note2));
assertThat(resultForQuery("test").size()).isEqualTo(3);
//when
// when
Paragraph p1 = note1.getLastParagraph();
p1.setText("no no no");
note1.persist(anonymous);
//then
// then
assertThat(resultForQuery("Notebook1").size()).isEqualTo(1);
List<Map<String, String>> results = resultForQuery("test");
assertThat(results).isNotEmpty();
assertThat(results.size()).isEqualTo(2);
//does not include Notebook1's paragraph any more
for (Map<String, String> result: results) {
assertThat(result.get("id").startsWith(note1.getId())).isFalse();;
// does not include Notebook1's paragraph any more
for (Map<String, String> result : results) {
assertThat(result.get("id").startsWith(note1.getId())).isFalse();
;
}
}
@Test public void indexNoteNameUpdatedOnNoteSave() throws IOException {
//given: total 2 notebooks, 3 paragraphs
@Test
public void indexNoteNameUpdatedOnNoteSave() throws IOException {
// given: total 2 notebooks, 3 paragraphs
Note note1 = newNoteWithParagraph("Notebook1", "test");
Note note2 = newNoteWithParagraphs("Notebook2", "not test", "not test at all");
noteSearchService.addIndexDocs(Arrays.asList(note1, note2));
assertThat(resultForQuery("test").size()).isEqualTo(3);
//when
// when
note1.setName("NotebookN");
note1.persist(anonymous);
//then
// then
assertThat(resultForQuery("Notebook1")).isEmpty();
assertThat(resultForQuery("NotebookN")).isNotEmpty();
assertThat(resultForQuery("NotebookN").size()).isEqualTo(1);
@ -246,8 +256,7 @@ public class LuceneSearchTest {
}
/**
* Creates a new Note \w given name,
* adds a new paragraph \w given text
* Creates a new Note \w given name, adds a new paragraph \w given text
*
* @param noteName name of the note
* @param parText text of the paragraph
@ -259,16 +268,13 @@ public class LuceneSearchTest {
return note1;
}
private Note newNoteWithParagraph(String noteName, String parText,String title) {
private Note newNoteWithParagraph(String noteName, String parText, String title) {
Note note = newNote(noteName);
addParagraphWithTextAndTitle(note, parText, title);
return note;
}
/**
* Creates a new Note \w given name,
* adds N paragraphs \w given texts
*/
/** Creates a new Note \w given name, adds N paragraphs \w given texts */
private Note newNoteWithParagraphs(String noteName, String... parTexts) {
Note note1 = newNote(noteName);
for (String parText : parTexts) {
@ -291,9 +297,16 @@ public class LuceneSearchTest {
}
private Note newNote(String name) {
Note note = new Note(notebookRepoMock, interpreterFactory, interpreterSettingManager, null, noteSearchService, null, null);
Note note =
new Note(
notebookRepoMock,
interpreterFactory,
interpreterSettingManager,
null,
noteSearchService,
null,
null);
note.setName(name);
return note;
}
}