Cancel
diff --git a/zeppelin-web/src/app/tabledata/pivot.js b/zeppelin-web/src/app/tabledata/pivot.js
index 1c938ea828..da2990043b 100644
--- a/zeppelin-web/src/app/tabledata/pivot.js
+++ b/zeppelin-web/src/app/tabledata/pivot.js
@@ -89,6 +89,7 @@ export default class PivotTransformation extends Transformation {
for (let j = i + 1; j < list.length; j++) {
if (angular.equals(list[i], list[j])) {
list.splice(j, 1)
+ j--
}
}
}
@@ -137,8 +138,8 @@ export default class PivotTransformation extends Transformation {
pivot (data, keys, groups, values) {
let aggrFunc = {
sum: function (a, b) {
- let varA = (a !== undefined) ? (isNaN(a) ? 1 : parseFloat(a)) : 0
- let varB = (b !== undefined) ? (isNaN(b) ? 1 : parseFloat(b)) : 0
+ let varA = (a !== undefined) ? (isNaN(a) ? 0 : parseFloat(a)) : 0
+ let varB = (b !== undefined) ? (isNaN(b) ? 0 : parseFloat(b)) : 0
return varA + varB
},
count: function (a, b) {
@@ -147,22 +148,38 @@ export default class PivotTransformation extends Transformation {
return varA + varB
},
min: function (a, b) {
- let varA = (a !== undefined) ? (isNaN(a) ? 1 : parseFloat(a)) : 0
- let varB = (b !== undefined) ? (isNaN(b) ? 1 : parseFloat(b)) : 0
- return Math.min(varA, varB)
+ let aIsValid = isValidNumber(a)
+ let bIsValid = isValidNumber(b)
+ if (!aIsValid) {
+ return parseFloat(b)
+ } else if (!bIsValid) {
+ return parseFloat(a)
+ } else {
+ return Math.min(parseFloat(a), parseFloat(b))
+ }
},
max: function (a, b) {
- let varA = (a !== undefined) ? (isNaN(a) ? 1 : parseFloat(a)) : 0
- let varB = (b !== undefined) ? (isNaN(b) ? 1 : parseFloat(b)) : 0
- return Math.max(varA, varB)
+ let aIsValid = isValidNumber(a)
+ let bIsValid = isValidNumber(b)
+ if (!aIsValid) {
+ return parseFloat(b)
+ } else if (!bIsValid) {
+ return parseFloat(a)
+ } else {
+ return Math.max(parseFloat(a), parseFloat(b))
+ }
},
avg: function (a, b, c) {
- let varA = (a !== undefined) ? (isNaN(a) ? 1 : parseFloat(a)) : 0
- let varB = (b !== undefined) ? (isNaN(b) ? 1 : parseFloat(b)) : 0
+ let varA = (a !== undefined) ? (isNaN(a) ? 0 : parseFloat(a)) : 0
+ let varB = (b !== undefined) ? (isNaN(b) ? 0 : parseFloat(b)) : 0
return varA + varB
}
}
+ let isValidNumber = function(num) {
+ return num !== undefined && !isNaN(num)
+ }
+
let aggrFuncDiv = {
sum: false,
count: false,
diff --git a/zeppelin-web/src/app/tabledata/pivot_settings.html b/zeppelin-web/src/app/tabledata/pivot_settings.html
index abfe73069b..68de019c27 100644
--- a/zeppelin-web/src/app/tabledata/pivot_settings.html
+++ b/zeppelin-web/src/app/tabledata/pivot_settings.html
@@ -49,7 +49,11 @@ limitations under the License.
data-drop="true" jqyoui-droppable="{multiple:true, onDrop:'save()'}"
class="list-unstyled"
style="border-radius: 6px; margin-top: 7px;">
-
+
{{item.name}}
@@ -69,7 +73,11 @@ limitations under the License.
jqyoui-droppable="{multiple:true, onDrop:'save()'}"
class="list-unstyled"
style="border-radius: 6px; margin-top: 7px;">
-
+
{{item.name}}
@@ -89,7 +97,11 @@ limitations under the License.
jqyoui-droppable="{multiple:true, onDrop:'save()'}"
class="list-unstyled"
style="border-radius: 6px; margin-top: 7px;">
-
+
+
diff --git a/zeppelin-web/src/app/tabledata/tabledata.test.js b/zeppelin-web/src/app/tabledata/tabledata.test.js
index 7e41de4bec..e24b073392 100644
--- a/zeppelin-web/src/app/tabledata/tabledata.test.js
+++ b/zeppelin-web/src/app/tabledata/tabledata.test.js
@@ -13,6 +13,7 @@
*/
import TableData from './tabledata.js'
+import PivotTransformation from './pivot.js'
describe('TableData build', function () {
let td
@@ -39,3 +40,92 @@ describe('TableData build', function () {
expect(td.comment).toBe('hello')
})
})
+
+describe('PivotTransformation build', function() {
+ let pt
+
+ beforeEach(function () {
+ console.log(PivotTransformation)
+ pt = new PivotTransformation()
+ })
+
+ it('check the result of keys, groups and values unique', function() {
+ // set inited mock data
+ let config = {
+ common: {
+ pivot: {
+ keys: [{index: 4, name: '4'},
+ {index: 3, name: '3'},
+ {index: 4, name: '4'},
+ {index: 3, name: '3'},
+ {index: 3, name: '3'},
+ {index: 3, name: '3'},
+ {index: 3, name: '3'},
+ {index: 5, name: '5'}],
+ groups: [],
+ values: []
+ }
+ }
+ }
+ pt.tableDataColumns = [
+ {index: 1, name: '1'},
+ {index: 2, name: '2'},
+ {index: 3, name: '3'},
+ {index: 4, name: '4'},
+ {index: 5, name: '5'}]
+
+ pt.setConfig(config)
+
+ pt.removeUnknown()
+
+ expect(config.common.pivot.keys.length).toBe(3)
+ expect(config.common.pivot.keys[0].index).toBe(4)
+ expect(config.common.pivot.keys[1].index).toBe(3)
+ expect(config.common.pivot.keys[2].index).toBe(5)
+ })
+
+ it('should aggregate values correctly', function() {
+ let td = new TableData()
+ td.loadParagraphResult({
+ type: 'TABLE',
+ msg: 'key\tvalue\na\t10\na\tnull\na\t0\na\t1\n'
+ })
+
+ let config = {
+ common: {
+ pivot: {
+ keys: [
+ {
+ 'name': 'key',
+ 'index': 0.0,
+ }
+ ],
+ groups: [],
+ values: [
+ {
+ 'name': 'value',
+ 'index': 1.0,
+ 'aggr': 'sum'
+ }
+ ]
+ }
+ }
+ }
+
+ pt.setConfig(config)
+ let transformed = pt.transform(td)
+ expect(transformed.rows['a']['value(sum)'].value).toBe(11)
+
+ pt.config.common.pivot.values[0].aggr = 'max'
+ transformed = pt.transform(td)
+ expect(transformed.rows['a']['value(max)'].value).toBe(10)
+
+ pt.config.common.pivot.values[0].aggr = 'min'
+ transformed = pt.transform(td)
+ expect(transformed.rows['a']['value(min)'].value).toBe(0)
+
+ pt.config.common.pivot.values[0].aggr = 'count'
+ transformed = pt.transform(td)
+ expect(transformed.rows['a']['value(count)'].value).toBe(4)
+ })
+})
diff --git a/zeppelin-web/src/components/login/login.html b/zeppelin-web/src/components/login/login.html
index d4a94a6f85..fd98bc8c4e 100644
--- a/zeppelin-web/src/components/login/login.html
+++ b/zeppelin-web/src/components/login/login.html
@@ -29,6 +29,7 @@ limitations under the License.
User Name
diff --git a/zeppelin-zengine/pom.xml b/zeppelin-zengine/pom.xml
index d614381845..ced7c39534 100644
--- a/zeppelin-zengine/pom.xml
+++ b/zeppelin-zengine/pom.xml
@@ -46,7 +46,7 @@
5.3.1
0.9.8
1.4.01
- 4.1.1.201511131810-r
+ 4.5.4.201711221230-r
1.3
diff --git a/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/ConfInterpreter.java b/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/ConfInterpreter.java
new file mode 100644
index 0000000000..d50c57b4da
--- /dev/null
+++ b/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/ConfInterpreter.java
@@ -0,0 +1,92 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.zeppelin.interpreter;
+
+import org.apache.commons.lang.exception.ExceptionUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.io.StringReader;
+import java.util.Properties;
+
+/**
+ * Special Interpreter for Interpreter Configuration customization. It is attached to each
+ * InterpreterGroup implicitly by Zeppelin.
+ */
+public class ConfInterpreter extends Interpreter {
+
+ private static Logger LOGGER = LoggerFactory.getLogger(ConfInterpreter.class);
+
+ private String interpreterGroupId;
+ private InterpreterSetting interpreterSetting;
+
+
+ public ConfInterpreter(Properties properties,
+ String interpreterGroupId,
+ InterpreterSetting interpreterSetting) {
+ super(properties);
+ this.interpreterGroupId = interpreterGroupId;
+ this.interpreterSetting = interpreterSetting;
+ }
+
+ @Override
+ public void open() throws InterpreterException {
+
+ }
+
+ @Override
+ public void close() throws InterpreterException {
+
+ }
+
+ @Override
+ public InterpreterResult interpret(String st, InterpreterContext context)
+ throws InterpreterException {
+
+ try {
+ Properties finalProperties = new Properties();
+ finalProperties.putAll(getProperties());
+ Properties newProperties = new Properties();
+ newProperties.load(new StringReader(st));
+ finalProperties.putAll(newProperties);
+ LOGGER.debug("Properties for InterpreterGroup: " + interpreterGroupId + " is "
+ + finalProperties);
+ interpreterSetting.setInterpreterGroupProperties(interpreterGroupId, finalProperties);
+ return new InterpreterResult(InterpreterResult.Code.SUCCESS);
+ } catch (IOException e) {
+ LOGGER.error("Fail to update interpreter setting", e);
+ return new InterpreterResult(InterpreterResult.Code.ERROR, ExceptionUtils.getStackTrace(e));
+ }
+ }
+
+ @Override
+ public void cancel(InterpreterContext context) throws InterpreterException {
+
+ }
+
+ @Override
+ public FormType getFormType() throws InterpreterException {
+ return null;
+ }
+
+ @Override
+ public int getProgress(InterpreterContext context) throws InterpreterException {
+ return 0;
+ }
+}
diff --git a/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/InterpreterSetting.java b/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/InterpreterSetting.java
index 26fcd8e93d..424aa27a16 100644
--- a/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/InterpreterSetting.java
+++ b/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/InterpreterSetting.java
@@ -38,6 +38,8 @@ import org.apache.zeppelin.interpreter.launcher.InterpreterLauncher;
import org.apache.zeppelin.interpreter.launcher.ShellScriptLauncher;
import org.apache.zeppelin.interpreter.launcher.SparkInterpreterLauncher;
import org.apache.zeppelin.interpreter.lifecycle.NullLifecycleManager;
+import org.apache.zeppelin.interpreter.recovery.NullRecoveryStorage;
+import org.apache.zeppelin.interpreter.recovery.RecoveryStorage;
import org.apache.zeppelin.interpreter.remote.RemoteAngularObjectRegistry;
import org.apache.zeppelin.interpreter.remote.RemoteInterpreter;
import org.apache.zeppelin.interpreter.remote.RemoteInterpreterEventPoller;
@@ -138,9 +140,14 @@ public class InterpreterSetting {
// launcher in future when we have other launcher implementation. e.g. third party launcher
// service like livy
private transient InterpreterLauncher launcher;
- ///////////////////////////////////////////////////////////////////////////////////////////
private transient LifecycleManager lifecycleManager;
+ ///////////////////////////////////////////////////////////////////////////////////////////
+
+
+
+ private transient RecoveryStorage recoveryStorage;
+ ///////////////////////////////////////////////////////////////////////////////////////////
/**
* Builder class for InterpreterSetting
@@ -240,6 +247,11 @@ public class InterpreterSetting {
return this;
}
+ public Builder setRecoveryStorage(RecoveryStorage recoveryStorage) {
+ interpreterSetting.recoveryStorage = recoveryStorage;
+ return this;
+ }
+
public InterpreterSetting create() {
// post processing
interpreterSetting.postProcessing();
@@ -259,6 +271,13 @@ public class InterpreterSetting {
if (this.lifecycleManager == null) {
this.lifecycleManager = new NullLifecycleManager(conf);
}
+ if (this.recoveryStorage == null) {
+ try {
+ this.recoveryStorage = new NullRecoveryStorage(conf, interpreterSettingManager);
+ } catch (IOException e) {
+ // ignore this exception as NullRecoveryStorage will do nothing.
+ }
+ }
}
/**
@@ -283,9 +302,9 @@ public class InterpreterSetting {
private void createLauncher() {
if (group.equals("spark")) {
- this.launcher = new SparkInterpreterLauncher(this.conf);
+ this.launcher = new SparkInterpreterLauncher(this.conf, this.recoveryStorage);
} else {
- this.launcher = new ShellScriptLauncher(this.conf);
+ this.launcher = new ShellScriptLauncher(this.conf, this.recoveryStorage);
}
}
@@ -342,6 +361,15 @@ public class InterpreterSetting {
return this;
}
+ public InterpreterSetting setRecoveryStorage(RecoveryStorage recoveryStorage) {
+ this.recoveryStorage = recoveryStorage;
+ return this;
+ }
+
+ public RecoveryStorage getRecoveryStorage() {
+ return recoveryStorage;
+ }
+
public LifecycleManager getLifecycleManager() {
return lifecycleManager;
}
@@ -406,7 +434,12 @@ public class InterpreterSetting {
}
void removeInterpreterGroup(String groupId) {
- this.interpreterGroups.remove(groupId);
+ try {
+ interpreterGroupWriteLock.lock();
+ this.interpreterGroups.remove(groupId);
+ } finally {
+ interpreterGroupWriteLock.unlock();
+ }
}
public ManagedInterpreterGroup getInterpreterGroup(String user, String noteId) {
@@ -423,7 +456,6 @@ public class InterpreterSetting {
return interpreterGroups.get(groupId);
}
- @VisibleForTesting
public ArrayList getAllInterpreterGroups() {
try {
interpreterGroupReadLock.lock();
@@ -648,12 +680,11 @@ public class InterpreterSetting {
///////////////////////////////////////////////////////////////////////////////////////
// This is the only place to create interpreters. For now we always create multiple interpreter
// together (one session). We don't support to create single interpreter yet.
- List createInterpreters(String user, String sessionId) {
+ List createInterpreters(String user, String interpreterGroupId, String sessionId) {
List interpreters = new ArrayList<>();
List interpreterInfos = getInterpreterInfos();
for (InterpreterInfo info : interpreterInfos) {
- Interpreter interpreter = null;
- interpreter = new RemoteInterpreter(getJavaProperties(), sessionId,
+ Interpreter interpreter = new RemoteInterpreter(getJavaProperties(), sessionId,
info.getClassName(), user, lifecycleManager);
if (info.isDefaultInterpreter()) {
interpreters.add(0, interpreter);
@@ -663,18 +694,23 @@ public class InterpreterSetting {
LOGGER.info("Interpreter {} created for user: {}, sessionId: {}",
interpreter.getClassName(), user, sessionId);
}
+ interpreters.add(new ConfInterpreter(getJavaProperties(), interpreterGroupId, this));
return interpreters;
}
- synchronized RemoteInterpreterProcess createInterpreterProcess() throws IOException {
+ synchronized RemoteInterpreterProcess createInterpreterProcess(String interpreterGroupId,
+ Properties properties)
+ throws IOException {
if (launcher == null) {
createLauncher();
}
InterpreterLaunchContext launchContext = new
- InterpreterLaunchContext(getJavaProperties(), option, interpreterRunner, id, group, name);
+ InterpreterLaunchContext(properties, option, interpreterRunner,
+ interpreterGroupId, id, group, name);
RemoteInterpreterProcess process = (RemoteInterpreterProcess) launcher.launch(launchContext);
process.setRemoteInterpreterEventPoller(
new RemoteInterpreterEventPoller(remoteInterpreterProcessListener, appEventListener));
+ recoveryStorage.onInterpreterClientStart(process);
return process;
}
@@ -716,6 +752,11 @@ public class InterpreterSetting {
return info.getClassName();
}
}
+ //TODO(zjffdu) It requires user can not create interpreter with name `conf`,
+ // conf is a reserved word of interpreter name
+ if (replName.equals("conf")) {
+ return ConfInterpreter.class.getName();
+ }
return null;
}
@@ -728,6 +769,29 @@ public class InterpreterSetting {
return interpreterGroup;
}
+ /**
+ * Throw exception when interpreter process has already launched
+ *
+ * @param interpreterGroupId
+ * @param properties
+ * @throws IOException
+ */
+ public void setInterpreterGroupProperties(String interpreterGroupId, Properties properties)
+ throws IOException {
+ ManagedInterpreterGroup interpreterGroup = this.interpreterGroups.get(interpreterGroupId);
+ for (List session : interpreterGroup.sessions.values()) {
+ for (Interpreter intp : session) {
+ if (!intp.getProperties().equals(properties) &&
+ interpreterGroup.getRemoteInterpreterProcess() != null &&
+ interpreterGroup.getRemoteInterpreterProcess().isRunning()) {
+ throw new IOException("Can not change interpreter properties when interpreter process " +
+ "has already been launched");
+ }
+ intp.setProperties(properties);
+ }
+ }
+ }
+
private void loadInterpreterDependencies() {
setStatus(Status.DOWNLOADING_DEPENDENCIES);
setErrorReason(null);
diff --git a/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/InterpreterSettingManager.java b/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/InterpreterSettingManager.java
index 0b7efd5db6..42f82fad21 100644
--- a/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/InterpreterSettingManager.java
+++ b/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/InterpreterSettingManager.java
@@ -34,12 +34,16 @@ import org.apache.zeppelin.dep.DependencyResolver;
import org.apache.zeppelin.display.AngularObjectRegistryListener;
import org.apache.zeppelin.helium.ApplicationEventListener;
import org.apache.zeppelin.interpreter.Interpreter.RegisteredInterpreter;
+import org.apache.zeppelin.interpreter.recovery.FileSystemRecoveryStorage;
+import org.apache.zeppelin.interpreter.recovery.NullRecoveryStorage;
+import org.apache.zeppelin.interpreter.recovery.RecoveryStorage;
import org.apache.zeppelin.interpreter.remote.RemoteInterpreterProcess;
import org.apache.zeppelin.interpreter.remote.RemoteInterpreterProcessListener;
import org.apache.zeppelin.interpreter.thrift.RemoteInterpreterService;
import org.apache.zeppelin.resource.Resource;
import org.apache.zeppelin.resource.ResourcePool;
import org.apache.zeppelin.resource.ResourceSet;
+import org.apache.zeppelin.util.ReflectionUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.sonatype.aether.repository.Authentication;
@@ -118,6 +122,7 @@ public class InterpreterSettingManager {
private ApplicationEventListener appEventListener;
private DependencyResolver dependencyResolver;
private LifecycleManager lifecycleManager;
+ private RecoveryStorage recoveryStorage;
public InterpreterSettingManager(ZeppelinConfiguration zeppelinConfiguration,
AngularObjectRegistryListener angularObjectRegistryListener,
@@ -154,13 +159,17 @@ public class InterpreterSettingManager {
this.angularObjectRegistryListener = angularObjectRegistryListener;
this.remoteInterpreterProcessListener = remoteInterpreterProcessListener;
this.appEventListener = appEventListener;
- try {
- this.lifecycleManager = (LifecycleManager)
- Class.forName(conf.getLifecycleManagerClass()).getConstructor(ZeppelinConfiguration.class)
- .newInstance(conf);
- } catch (Exception e) {
- throw new IOException("Fail to create LifecycleManager", e);
- }
+
+ this.recoveryStorage = ReflectionUtils.createClazzInstance(conf.getRecoveryStorageClass(),
+ new Class[] {ZeppelinConfiguration.class, InterpreterSettingManager.class},
+ new Object[] {conf, this});
+ this.recoveryStorage.init();
+ LOGGER.info("Using RecoveryStorage: " + this.recoveryStorage.getClass().getName());
+
+ this.lifecycleManager = ReflectionUtils.createClazzInstance(conf.getLifecycleManagerClass(),
+ new Class[] {ZeppelinConfiguration.class},
+ new Object[] {conf});
+ LOGGER.info("Using LifecycleManager: " + this.lifecycleManager.getClass().getName());
init();
}
@@ -174,6 +183,7 @@ public class InterpreterSettingManager {
.setAppEventListener(appEventListener)
.setDependencyResolver(dependencyResolver)
.setLifecycleManager(lifecycleManager)
+ .setRecoveryStorage(recoveryStorage)
.postProcessing();
}
@@ -307,8 +317,16 @@ public class InterpreterSettingManager {
saveToFile();
}
+ public RemoteInterpreterProcessListener getRemoteInterpreterProcessListener() {
+ return remoteInterpreterProcessListener;
+ }
+
+ public ApplicationEventListener getAppEventListener() {
+ return appEventListener;
+ }
+
private boolean registerInterpreterFromResource(ClassLoader cl, String interpreterDir,
- String interpreterJson) throws IOException {
+ String interpreterJson) throws IOException {
URL[] urls = recursiveBuildLibList(new File(interpreterDir));
ClassLoader tempClassLoader = new URLClassLoader(urls, null);
@@ -507,6 +525,10 @@ public class InterpreterSettingManager {
return resourceSet;
}
+ public RecoveryStorage getRecoveryStorage() {
+ return recoveryStorage;
+ }
+
public void removeResourcesBelongsToParagraph(String noteId, String paragraphId) {
for (ManagedInterpreterGroup intpGroup : getAllInterpreterGroup()) {
ResourceSet resourceSet = new ResourceSet();
diff --git a/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/ManagedInterpreterGroup.java b/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/ManagedInterpreterGroup.java
index 219204f041..641c0ac23e 100644
--- a/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/ManagedInterpreterGroup.java
+++ b/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/ManagedInterpreterGroup.java
@@ -28,6 +28,7 @@ import org.slf4j.LoggerFactory;
import java.io.IOException;
import java.util.Collection;
import java.util.List;
+import java.util.Properties;
/**
* ManagedInterpreterGroup runs under zeppelin server
@@ -54,14 +55,31 @@ public class ManagedInterpreterGroup extends InterpreterGroup {
return interpreterSetting;
}
- public synchronized RemoteInterpreterProcess getOrCreateInterpreterProcess() throws IOException {
+ public synchronized RemoteInterpreterProcess getOrCreateInterpreterProcess(String userName,
+ Properties properties)
+ throws IOException {
if (remoteInterpreterProcess == null) {
LOGGER.info("Create InterpreterProcess for InterpreterGroup: " + getId());
- remoteInterpreterProcess = interpreterSetting.createInterpreterProcess();
+ remoteInterpreterProcess = interpreterSetting.createInterpreterProcess(id, properties);
+ synchronized (remoteInterpreterProcess) {
+ if (!remoteInterpreterProcess.isRunning()) {
+ remoteInterpreterProcess.start(userName, false);
+ remoteInterpreterProcess.getRemoteInterpreterEventPoller()
+ .setInterpreterProcess(remoteInterpreterProcess);
+ remoteInterpreterProcess.getRemoteInterpreterEventPoller().setInterpreterGroup(this);
+ remoteInterpreterProcess.getRemoteInterpreterEventPoller().start();
+ getInterpreterSetting().getRecoveryStorage()
+ .onInterpreterClientStart(remoteInterpreterProcess);
+ }
+ }
}
return remoteInterpreterProcess;
}
+ public RemoteInterpreterProcess getInterpreterProcess() {
+ return remoteInterpreterProcess;
+ }
+
public RemoteInterpreterProcess getRemoteInterpreterProcess() {
return remoteInterpreterProcess;
}
@@ -92,6 +110,11 @@ public class ManagedInterpreterGroup extends InterpreterGroup {
if (remoteInterpreterProcess != null) {
LOGGER.info("Kill RemoteInterpreterProcess");
remoteInterpreterProcess.stop();
+ try {
+ interpreterSetting.getRecoveryStorage().onInterpreterClientStop(remoteInterpreterProcess);
+ } catch (IOException e) {
+ LOGGER.error("Fail to store recovery data", e);
+ }
remoteInterpreterProcess = null;
}
}
@@ -131,7 +154,7 @@ public class ManagedInterpreterGroup extends InterpreterGroup {
if (sessions.containsKey(sessionId)) {
return sessions.get(sessionId);
} else {
- List interpreters = interpreterSetting.createInterpreters(user, sessionId);
+ List interpreters = interpreterSetting.createInterpreters(user, id, sessionId);
for (Interpreter interpreter : interpreters) {
interpreter.setInterpreterGroup(this);
}
diff --git a/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/launcher/ShellScriptLauncher.java b/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/launcher/ShellScriptLauncher.java
index 8c86129f64..6ddcacf275 100644
--- a/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/launcher/ShellScriptLauncher.java
+++ b/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/launcher/ShellScriptLauncher.java
@@ -21,50 +21,68 @@ package org.apache.zeppelin.interpreter.launcher;
import org.apache.zeppelin.conf.ZeppelinConfiguration;
import org.apache.zeppelin.interpreter.InterpreterOption;
import org.apache.zeppelin.interpreter.InterpreterRunner;
+import org.apache.zeppelin.interpreter.recovery.RecoveryStorage;
import org.apache.zeppelin.interpreter.remote.RemoteInterpreterManagedProcess;
import org.apache.zeppelin.interpreter.remote.RemoteInterpreterRunningProcess;
import org.apache.zeppelin.interpreter.remote.RemoteInterpreterUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
/**
* Interpreter Launcher which use shell script to launch the interpreter process.
- *
*/
public class ShellScriptLauncher extends InterpreterLauncher {
private static final Logger LOGGER = LoggerFactory.getLogger(ShellScriptLauncher.class);
- public ShellScriptLauncher(ZeppelinConfiguration zConf) {
- super(zConf);
+ public ShellScriptLauncher(ZeppelinConfiguration zConf, RecoveryStorage recoveryStorage) {
+ super(zConf, recoveryStorage);
}
@Override
- public InterpreterClient launch(InterpreterLaunchContext context) {
+ public InterpreterClient launch(InterpreterLaunchContext context) throws IOException {
LOGGER.info("Launching Interpreter: " + context.getInterpreterSettingGroup());
this.properties = context.getProperties();
InterpreterOption option = context.getOption();
InterpreterRunner runner = context.getRunner();
String groupName = context.getInterpreterSettingGroup();
String name = context.getInterpreterSettingName();
-
int connectTimeout =
zConf.getInt(ZeppelinConfiguration.ConfVars.ZEPPELIN_INTERPRETER_CONNECT_TIMEOUT);
+
if (option.isExistingProcess()) {
return new RemoteInterpreterRunningProcess(
+ context.getInterpreterSettingName(),
connectTimeout,
option.getHost(),
option.getPort());
} else {
+ // try to recover it first
+ if (zConf.isRecoveryEnabled()) {
+ InterpreterClient recoveredClient =
+ recoveryStorage.getInterpreterClient(context.getInterpreterGroupId());
+ if (recoveredClient != null) {
+ if (recoveredClient.isRunning()) {
+ LOGGER.info("Recover interpreter process: " + recoveredClient.getHost() + ":" +
+ recoveredClient.getPort());
+ return recoveredClient;
+ } else {
+ LOGGER.warn("Cannot recover interpreter process: " + recoveredClient.getHost() + ":"
+ + recoveredClient.getPort() + ", as it is already terminated.");
+ }
+ }
+ }
+
// create new remote process
String localRepoPath = zConf.getInterpreterLocalRepoPath() + "/"
+ context.getInterpreterSettingId();
return new RemoteInterpreterManagedProcess(
runner != null ? runner.getPath() : zConf.getInterpreterRemoteRunnerPath(),
- zConf.getCallbackPortRange(), zConf.getInterpreterPortRange(),
+ zConf.getCallbackPortRange(), zConf.getInterpreterPortRange(),
zConf.getInterpreterDir() + "/" + groupName, localRepoPath,
buildEnvFromProperties(), connectTimeout, name);
}
diff --git a/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/launcher/SparkInterpreterLauncher.java b/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/launcher/SparkInterpreterLauncher.java
index 32a0530af1..e8a9cdf881 100644
--- a/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/launcher/SparkInterpreterLauncher.java
+++ b/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/launcher/SparkInterpreterLauncher.java
@@ -19,6 +19,7 @@ package org.apache.zeppelin.interpreter.launcher;
import org.apache.commons.lang3.StringUtils;
import org.apache.zeppelin.conf.ZeppelinConfiguration;
+import org.apache.zeppelin.interpreter.recovery.RecoveryStorage;
import org.apache.zeppelin.interpreter.remote.RemoteInterpreterUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -35,8 +36,8 @@ public class SparkInterpreterLauncher extends ShellScriptLauncher {
private static final Logger LOGGER = LoggerFactory.getLogger(SparkInterpreterLauncher.class);
- public SparkInterpreterLauncher(ZeppelinConfiguration zConf) {
- super(zConf);
+ public SparkInterpreterLauncher(ZeppelinConfiguration zConf, RecoveryStorage recoveryStorage) {
+ super(zConf, recoveryStorage);
}
@Override
diff --git a/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/recovery/FileSystemRecoveryStorage.java b/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/recovery/FileSystemRecoveryStorage.java
new file mode 100644
index 0000000000..5a0c8adf6c
--- /dev/null
+++ b/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/recovery/FileSystemRecoveryStorage.java
@@ -0,0 +1,139 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.zeppelin.interpreter.recovery;
+
+import org.apache.commons.lang.StringUtils;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.PathFilter;
+import org.apache.hadoop.io.IOUtils;
+import org.apache.hadoop.security.UserGroupInformation;
+import org.apache.zeppelin.conf.ZeppelinConfiguration;
+import org.apache.zeppelin.interpreter.InterpreterSetting;
+import org.apache.zeppelin.interpreter.InterpreterSettingManager;
+import org.apache.zeppelin.interpreter.ManagedInterpreterGroup;
+import org.apache.zeppelin.interpreter.launcher.InterpreterClient;
+import org.apache.zeppelin.interpreter.remote.RemoteInterpreterEventPoller;
+import org.apache.zeppelin.interpreter.remote.RemoteInterpreterProcess;
+import org.apache.zeppelin.interpreter.remote.RemoteInterpreterRunningProcess;
+import org.apache.zeppelin.notebook.FileSystemStorage;
+import org.apache.zeppelin.notebook.repo.FileSystemNotebookRepo;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+
+/**
+ * Hadoop compatible FileSystem based RecoveryStorage implementation.
+ *
+ * Save InterpreterProcess in the format of:
+ * InterpreterGroupId host:port
+ */
+public class FileSystemRecoveryStorage extends RecoveryStorage {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(FileSystemRecoveryStorage.class);
+
+ private InterpreterSettingManager interpreterSettingManager;
+ private FileSystemStorage fs;
+ private Path recoveryDir;
+
+ public FileSystemRecoveryStorage(ZeppelinConfiguration zConf,
+ InterpreterSettingManager interpreterSettingManager)
+ throws IOException {
+ super(zConf);
+ this.interpreterSettingManager = interpreterSettingManager;
+ this.zConf = zConf;
+ this.fs = FileSystemStorage.get(zConf);
+ this.recoveryDir = this.fs.makeQualified(new Path(zConf.getRecoveryDir()));
+ LOGGER.info("Using folder {} to store recovery data", recoveryDir);
+ this.fs.tryMkDir(recoveryDir);
+ }
+
+ @Override
+ public void onInterpreterClientStart(InterpreterClient client) throws IOException {
+ save(client.getInterpreterSettingName());
+ }
+
+ @Override
+ public void onInterpreterClientStop(InterpreterClient client) throws IOException {
+ save(client.getInterpreterSettingName());
+ }
+
+ private void save(String interpreterSettingName) throws IOException {
+ InterpreterSetting interpreterSetting =
+ interpreterSettingManager.getInterpreterSettingByName(interpreterSettingName);
+ List recoveryContent = new ArrayList<>();
+ for (ManagedInterpreterGroup interpreterGroup : interpreterSetting.getAllInterpreterGroups()) {
+ RemoteInterpreterProcess interpreterProcess = interpreterGroup.getInterpreterProcess();
+ if (interpreterProcess != null) {
+ recoveryContent.add(interpreterGroup.getId() + "\t" + interpreterProcess.getHost() + ":" +
+ interpreterProcess.getPort());
+ }
+ }
+ LOGGER.debug("Updating recovery data for interpreterSetting: " + interpreterSettingName);
+ LOGGER.debug("Recovery Data: " + StringUtils.join(recoveryContent, System.lineSeparator()));
+ Path recoveryFile = new Path(recoveryDir, interpreterSettingName + ".recovery");
+ fs.writeFile(StringUtils.join(recoveryContent, System.lineSeparator()), recoveryFile, true);
+ }
+
+ @Override
+ public Map restore() throws IOException {
+ Map clients = new HashMap<>();
+ List paths = fs.list(new Path(recoveryDir + "/*.recovery"));
+
+ for (Path path : paths) {
+ String fileName = path.getName();
+ String interpreterSettingName = fileName.substring(0,
+ fileName.length() - ".recovery".length());
+ String recoveryContent = fs.readFile(path);
+ if (!StringUtils.isBlank(recoveryContent)) {
+ for (String line : recoveryContent.split(System.lineSeparator())) {
+ String[] tokens = line.split("\t");
+ String groupId = tokens[0];
+ String[] hostPort = tokens[1].split(":");
+ int connectTimeout =
+ zConf.getInt(ZeppelinConfiguration.ConfVars.ZEPPELIN_INTERPRETER_CONNECT_TIMEOUT);
+ RemoteInterpreterRunningProcess client = new RemoteInterpreterRunningProcess(
+ interpreterSettingName, connectTimeout, hostPort[0], Integer.parseInt(hostPort[1]));
+ // interpreterSettingManager may be null when this class is used when it is used
+ // stop-interpreter.sh
+ if (interpreterSettingManager != null) {
+ client.setRemoteInterpreterEventPoller(new RemoteInterpreterEventPoller(
+ interpreterSettingManager.getRemoteInterpreterProcessListener(),
+ interpreterSettingManager.getAppEventListener()));
+ }
+ clients.put(groupId, client);
+ LOGGER.info("Recovering Interpreter Process: " + hostPort[0] + ":" + hostPort[1]);
+ }
+ }
+ }
+
+ return clients;
+ }
+}
diff --git a/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/recovery/NullRecoveryStorage.java b/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/recovery/NullRecoveryStorage.java
new file mode 100644
index 0000000000..3a7d12c70f
--- /dev/null
+++ b/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/recovery/NullRecoveryStorage.java
@@ -0,0 +1,54 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.zeppelin.interpreter.recovery;
+
+import org.apache.zeppelin.conf.ZeppelinConfiguration;
+import org.apache.zeppelin.interpreter.InterpreterSettingManager;
+import org.apache.zeppelin.interpreter.launcher.InterpreterClient;
+
+import java.io.IOException;
+import java.util.Map;
+
+
+/**
+ * RecoveryStorage that do nothing, used when recovery is not enabled.
+ *
+ */
+public class NullRecoveryStorage extends RecoveryStorage {
+
+ public NullRecoveryStorage(ZeppelinConfiguration zConf,
+ InterpreterSettingManager interpreterSettingManager)
+ throws IOException {
+ super(zConf);
+ }
+
+ @Override
+ public void onInterpreterClientStart(InterpreterClient client) throws IOException {
+
+ }
+
+ @Override
+ public void onInterpreterClientStop(InterpreterClient client) throws IOException {
+
+ }
+
+ @Override
+ public Map restore() throws IOException {
+ return null;
+ }
+}
diff --git a/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/recovery/StopInterpreter.java b/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/recovery/StopInterpreter.java
new file mode 100644
index 0000000000..d74b1621e7
--- /dev/null
+++ b/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/recovery/StopInterpreter.java
@@ -0,0 +1,40 @@
+package org.apache.zeppelin.interpreter.recovery;
+
+import org.apache.zeppelin.conf.ZeppelinConfiguration;
+import org.apache.zeppelin.interpreter.InterpreterSettingManager;
+import org.apache.zeppelin.interpreter.launcher.InterpreterClient;
+import org.apache.zeppelin.util.ReflectionUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.Map;
+
+
+/**
+ * Utility class for stopping interpreter in the case that you want to stop all the
+ * interpreter process even when you enable recovery, or you want to kill interpreter process
+ * to avoid orphan process.
+ */
+public class StopInterpreter {
+
+ private static Logger LOGGER = LoggerFactory.getLogger(StopInterpreter.class);
+
+ public static void main(String[] args) throws IOException {
+ ZeppelinConfiguration zConf = ZeppelinConfiguration.create();
+ RecoveryStorage recoveryStorage = null;
+
+ recoveryStorage = ReflectionUtils.createClazzInstance(zConf.getRecoveryStorageClass(),
+ new Class[] {ZeppelinConfiguration.class, InterpreterSettingManager.class},
+ new Object[] {zConf, null});
+
+ LOGGER.info("Using RecoveryStorage: " + recoveryStorage.getClass().getName());
+ Map restoredClients = recoveryStorage.restore();
+ if (restoredClients != null) {
+ for (InterpreterClient client : restoredClients.values()) {
+ LOGGER.info("Stop Interpreter Process: " + client.getHost() + ":" + client.getPort());
+ client.stop();
+ }
+ }
+ }
+}
diff --git a/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreter.java b/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreter.java
index 8964210c1c..bda8010d93 100644
--- a/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreter.java
+++ b/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreter.java
@@ -25,6 +25,7 @@ import org.apache.zeppelin.display.AngularObject;
import org.apache.zeppelin.display.AngularObjectRegistry;
import org.apache.zeppelin.display.GUI;
import org.apache.zeppelin.display.Input;
+import org.apache.zeppelin.interpreter.ConfInterpreter;
import org.apache.zeppelin.interpreter.Interpreter;
import org.apache.zeppelin.interpreter.InterpreterContext;
import org.apache.zeppelin.interpreter.InterpreterContextRunner;
@@ -101,16 +102,7 @@ public class RemoteInterpreter extends Interpreter {
return this.interpreterProcess;
}
ManagedInterpreterGroup intpGroup = getInterpreterGroup();
- this.interpreterProcess = intpGroup.getOrCreateInterpreterProcess();
- synchronized (interpreterProcess) {
- if (!interpreterProcess.isRunning()) {
- interpreterProcess.start(this.getUserName(), false);
- interpreterProcess.getRemoteInterpreterEventPoller()
- .setInterpreterProcess(interpreterProcess);
- interpreterProcess.getRemoteInterpreterEventPoller().setInterpreterGroup(intpGroup);
- interpreterProcess.getRemoteInterpreterEventPoller().start();
- }
- }
+ this.interpreterProcess = intpGroup.getOrCreateInterpreterProcess(getUserName(), properties);
return interpreterProcess;
}
@@ -130,7 +122,9 @@ public class RemoteInterpreter extends Interpreter {
for (Interpreter interpreter : getInterpreterGroup()
.getOrCreateSession(this.getUserName(), sessionId)) {
try {
- ((RemoteInterpreter) interpreter).internal_create();
+ if (!(interpreter instanceof ConfInterpreter)) {
+ ((RemoteInterpreter) interpreter).internal_create();
+ }
} catch (IOException e) {
throw new InterpreterException(e);
}
@@ -346,8 +340,7 @@ public class RemoteInterpreter extends Interpreter {
final InterpreterContext interpreterContext)
throws InterpreterException {
if (!isOpened) {
- LOGGER.warn("completion is called when RemoterInterpreter is not opened for " + className);
- return new ArrayList<>();
+ open();
}
RemoteInterpreterProcess interpreterProcess = null;
try {
diff --git a/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreterManagedProcess.java b/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreterManagedProcess.java
index 27e826c70a..3dd5bfa349 100644
--- a/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreterManagedProcess.java
+++ b/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreterManagedProcess.java
@@ -214,7 +214,7 @@ public class RemoteInterpreterManagedProcess extends RemoteInterpreterProcess
callbackServer.stop();
}
if (isRunning()) {
- logger.info("kill interpreter process");
+ logger.info("Kill interpreter process");
try {
callRemoteFunction(new RemoteFunction() {
@Override
@@ -263,7 +263,6 @@ public class RemoteInterpreterManagedProcess extends RemoteInterpreterProcess
return interpreterDir;
}
- @VisibleForTesting
public String getInterpreterSettingName() {
return interpreterSettingName;
}
diff --git a/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreterProcess.java b/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreterProcess.java
index 88cc4894be..08653ae390 100644
--- a/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreterProcess.java
+++ b/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreterProcess.java
@@ -51,12 +51,6 @@ public abstract class RemoteInterpreterProcess implements InterpreterClient {
this.remoteInterpreterEventPoller = eventPoller;
}
- public abstract String getHost();
- public abstract int getPort();
- public abstract void start(String userName, Boolean isUserImpersonate);
- public abstract void stop();
- public abstract boolean isRunning();
-
public int getConnectTimeout() {
return connectTimeout;
}
diff --git a/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreterRunningProcess.java b/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreterRunningProcess.java
index d8715a0d49..0e87e4f7d4 100644
--- a/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreterRunningProcess.java
+++ b/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreterRunningProcess.java
@@ -17,6 +17,7 @@
package org.apache.zeppelin.interpreter.remote;
import org.apache.zeppelin.helium.ApplicationEventListener;
+import org.apache.zeppelin.interpreter.thrift.RemoteInterpreterService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -27,13 +28,16 @@ public class RemoteInterpreterRunningProcess extends RemoteInterpreterProcess {
private final Logger logger = LoggerFactory.getLogger(RemoteInterpreterRunningProcess.class);
private final String host;
private final int port;
+ private final String interpreterSettingName;
public RemoteInterpreterRunningProcess(
+ String interpreterSettingName,
int connectTimeout,
String host,
int port
) {
super(connectTimeout);
+ this.interpreterSettingName = interpreterSettingName;
this.host = host;
this.port = port;
}
@@ -48,6 +52,11 @@ public class RemoteInterpreterRunningProcess extends RemoteInterpreterProcess {
return port;
}
+ @Override
+ public String getInterpreterSettingName() {
+ return interpreterSettingName;
+ }
+
@Override
public void start(String userName, Boolean isUserImpersonate) {
// assume process is externally managed. nothing to do
@@ -55,7 +64,24 @@ public class RemoteInterpreterRunningProcess extends RemoteInterpreterProcess {
@Override
public void stop() {
- // assume process is externally managed. nothing to do
+ // assume process is externally managed. nothing to do. But will kill it
+ // when you want to force stop it. ENV ZEPPELIN_FORCE_STOP control that.
+ if (System.getenv("ZEPPELIN_FORCE_STOP") != null) {
+ if (isRunning()) {
+ logger.info("Kill interpreter process");
+ try {
+ callRemoteFunction(new RemoteFunction() {
+ @Override
+ public Void call(RemoteInterpreterService.Client client) throws Exception {
+ client.shutdown();
+ return null;
+ }
+ });
+ } catch (Exception e) {
+ logger.warn("ignore the exception when shutting down interpreter process.", e);
+ }
+ }
+ }
}
@Override
diff --git a/zeppelin-zengine/src/main/java/org/apache/zeppelin/notebook/FileSystemStorage.java b/zeppelin-zengine/src/main/java/org/apache/zeppelin/notebook/FileSystemStorage.java
new file mode 100644
index 0000000000..6f3d3f97f5
--- /dev/null
+++ b/zeppelin-zengine/src/main/java/org/apache/zeppelin/notebook/FileSystemStorage.java
@@ -0,0 +1,168 @@
+package org.apache.zeppelin.notebook;
+
+import org.apache.commons.lang.StringUtils;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.RawLocalFileSystem;
+import org.apache.hadoop.io.IOUtils;
+import org.apache.hadoop.security.UserGroupInformation;
+import org.apache.zeppelin.conf.ZeppelinConfiguration;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.security.PrivilegedExceptionAction;
+import java.util.ArrayList;
+import java.util.List;
+
+
+/**
+ * Hadoop FileSystem wrapper. Support both secure and no-secure mode
+ */
+public class FileSystemStorage {
+
+ private static Logger LOGGER = LoggerFactory.getLogger(FileSystemStorage.class);
+
+ private static FileSystemStorage instance;
+
+ private ZeppelinConfiguration zConf;
+ private Configuration hadoopConf;
+ private boolean isSecurityEnabled = false;
+ private FileSystem fs;
+
+ private FileSystemStorage(ZeppelinConfiguration zConf) throws IOException {
+ this.zConf = zConf;
+ this.hadoopConf = new Configuration();
+ this.hadoopConf.set("fs.file.impl", RawLocalFileSystem.class.getName());
+ this.isSecurityEnabled = UserGroupInformation.isSecurityEnabled();
+
+ if (isSecurityEnabled) {
+ String keytab = zConf.getString(
+ ZeppelinConfiguration.ConfVars.ZEPPELIN_SERVER_KERBEROS_KEYTAB);
+ String principal = zConf.getString(
+ ZeppelinConfiguration.ConfVars.ZEPPELIN_SERVER_KERBEROS_PRINCIPAL);
+ if (StringUtils.isBlank(keytab) || StringUtils.isBlank(principal)) {
+ throw new IOException("keytab and principal can not be empty, keytab: " + keytab
+ + ", principal: " + principal);
+ }
+ UserGroupInformation.loginUserFromKeytab(principal, keytab);
+ }
+
+ try {
+ this.fs = FileSystem.get(new URI(zConf.getNotebookDir()), this.hadoopConf);
+ LOGGER.info("Creating FileSystem: " + this.fs.getClass().getCanonicalName());
+ } catch (URISyntaxException e) {
+ throw new IOException(e);
+ }
+ }
+
+ public static synchronized FileSystemStorage get(ZeppelinConfiguration zConf) throws IOException {
+ if (instance == null) {
+ instance = new FileSystemStorage(zConf);
+ }
+ return instance;
+ }
+
+ public Path makeQualified(Path path) {
+ return fs.makeQualified(path);
+ }
+
+ public void tryMkDir(final Path dir) throws IOException {
+ callHdfsOperation(new HdfsOperation() {
+ @Override
+ public Void call() throws IOException {
+ if (!fs.exists(dir)) {
+ fs.mkdirs(dir);
+ LOGGER.info("Create dir {} in hdfs", dir.toString());
+ }
+ if (fs.isFile(dir)) {
+ throw new IOException(dir.toString() + " is file instead of directory, please remove " +
+ "it or specify another directory");
+ }
+ fs.mkdirs(dir);
+ return null;
+ }
+ });
+ }
+
+ public List list(final Path path) throws IOException {
+ return callHdfsOperation(new HdfsOperation>() {
+ @Override
+ public List call() throws IOException {
+ List paths = new ArrayList<>();
+ for (FileStatus status : fs.globStatus(path)) {
+ paths.add(status.getPath());
+ }
+ return paths;
+ }
+ });
+ }
+
+ public boolean delete(final Path path) throws IOException {
+ return callHdfsOperation(new HdfsOperation() {
+ @Override
+ public Boolean call() throws IOException {
+ return fs.delete(path, true);
+ }
+ });
+ }
+
+ public String readFile(final Path file) throws IOException {
+ return callHdfsOperation(new HdfsOperation() {
+ @Override
+ public String call() throws IOException {
+ LOGGER.debug("Read from file: " + file);
+ ByteArrayOutputStream noteBytes = new ByteArrayOutputStream();
+ IOUtils.copyBytes(fs.open(file), noteBytes, hadoopConf);
+ return new String(noteBytes.toString(
+ zConf.getString(ZeppelinConfiguration.ConfVars.ZEPPELIN_ENCODING)));
+ }
+ });
+ }
+
+ public void writeFile(final String content, final Path file, boolean writeTempFileFirst)
+ throws IOException {
+ callHdfsOperation(new HdfsOperation() {
+ @Override
+ public Void call() throws IOException {
+ InputStream in = new ByteArrayInputStream(content.getBytes(
+ zConf.getString(ZeppelinConfiguration.ConfVars.ZEPPELIN_ENCODING)));
+ Path tmpFile = new Path(file.toString() + ".tmp");
+ IOUtils.copyBytes(in, fs.create(tmpFile), hadoopConf);
+ fs.delete(file, true);
+ fs.rename(tmpFile, file);
+ return null;
+ }
+ });
+ }
+
+ private interface HdfsOperation {
+ T call() throws IOException;
+ }
+
+ public synchronized T callHdfsOperation(final HdfsOperation func) throws IOException {
+ if (isSecurityEnabled) {
+ UserGroupInformation.getLoginUser().reloginFromKeytab();
+ try {
+ return UserGroupInformation.getCurrentUser().doAs(new PrivilegedExceptionAction() {
+ @Override
+ public T run() throws Exception {
+ return func.call();
+ }
+ });
+ } catch (InterruptedException e) {
+ throw new IOException(e);
+ }
+ } else {
+ return func.call();
+ }
+ }
+
+}
diff --git a/zeppelin-zengine/src/main/java/org/apache/zeppelin/notebook/Paragraph.java b/zeppelin-zengine/src/main/java/org/apache/zeppelin/notebook/Paragraph.java
index 10a8548798..32b9b73263 100644
--- a/zeppelin-zengine/src/main/java/org/apache/zeppelin/notebook/Paragraph.java
+++ b/zeppelin-zengine/src/main/java/org/apache/zeppelin/notebook/Paragraph.java
@@ -191,7 +191,7 @@ public class Paragraph extends Job implements Cloneable, JsonSerializable {
this.scriptText = this.text.substring(headingSpace.length() + intpText.length() + 1).trim();
} else {
this.intpText = "";
- this.scriptText = this.text;
+ this.scriptText = this.text.trim();
}
}
}
@@ -250,14 +250,17 @@ public class Paragraph extends Job implements Cloneable, JsonSerializable {
return note.getInterpreterCompletion();
}
}
- String trimmedBuffer = buffer != null ? buffer.trim() : null;
- cursor = calculateCursorPosition(buffer, trimmedBuffer, cursor);
+ this.interpreter = getBindedInterpreter();
+
+ setText(buffer);
+
+ cursor = calculateCursorPosition(buffer, cursor);
InterpreterContext interpreterContext = getInterpreterContextWithoutRunner(null);
try {
if (this.interpreter != null) {
- return this.interpreter.completion(scriptText, cursor, interpreterContext);
+ return this.interpreter.completion(this.scriptText, cursor, interpreterContext);
} else {
return null;
}
@@ -266,24 +269,15 @@ public class Paragraph extends Job implements Cloneable, JsonSerializable {
}
}
- public int calculateCursorPosition(String buffer, String trimmedBuffer, int cursor) {
- int countWhitespacesAtStart = buffer.indexOf(trimmedBuffer);
- if (countWhitespacesAtStart > 0) {
- cursor -= countWhitespacesAtStart;
- }
+ public int calculateCursorPosition(String buffer, int cursor) {
+ // scriptText trimmed
- // parse text to get interpreter component
- String repl = null;
- if (trimmedBuffer != null) {
- Matcher matcher = REPL_PATTERN.matcher(trimmedBuffer);
- if (matcher.matches()) {
- repl = matcher.group(2);
- }
+ if (this.scriptText.isEmpty()) {
+ return 0;
}
-
- if (repl != null && cursor > repl.length()) {
- String body = trimmedBuffer.substring(repl.length() + 1);
- cursor -= repl.length() + 1 + body.indexOf(body.trim());
+ int countCharactersBeforeScript = buffer.indexOf(this.scriptText);
+ if (countCharactersBeforeScript > 0) {
+ cursor -= countCharactersBeforeScript;
}
return cursor;
@@ -357,6 +351,7 @@ public class Paragraph extends Job implements Cloneable, JsonSerializable {
setStatus(Job.Status.ERROR);
throw intpException;
}
+ setStatus(Status.READY);
if (getConfig().get("enabled") == null || (Boolean) getConfig().get("enabled")) {
setAuthenticationInfo(getAuthenticationInfo());
interpreter.getScheduler().submit(this);
diff --git a/zeppelin-zengine/src/main/java/org/apache/zeppelin/notebook/repo/FileSystemNotebookRepo.java b/zeppelin-zengine/src/main/java/org/apache/zeppelin/notebook/repo/FileSystemNotebookRepo.java
index ba858e6925..d8ec0e5400 100644
--- a/zeppelin-zengine/src/main/java/org/apache/zeppelin/notebook/repo/FileSystemNotebookRepo.java
+++ b/zeppelin-zengine/src/main/java/org/apache/zeppelin/notebook/repo/FileSystemNotebookRepo.java
@@ -8,6 +8,7 @@ import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.security.UserGroupInformation;
import org.apache.zeppelin.conf.ZeppelinConfiguration;
+import org.apache.zeppelin.notebook.FileSystemStorage;
import org.apache.zeppelin.notebook.Note;
import org.apache.zeppelin.notebook.NoteInfo;
import org.apache.zeppelin.user.AuthenticationInfo;
@@ -37,108 +38,45 @@ import java.util.Map;
public class FileSystemNotebookRepo implements NotebookRepo {
private static final Logger LOGGER = LoggerFactory.getLogger(FileSystemNotebookRepo.class);
- private Configuration hadoopConf;
- private ZeppelinConfiguration zConf;
- private boolean isSecurityEnabled = false;
- private FileSystem fs;
+ private FileSystemStorage fs;
private Path notebookDir;
public FileSystemNotebookRepo(ZeppelinConfiguration zConf) throws IOException {
- this.zConf = zConf;
- this.hadoopConf = new Configuration();
+ this.fs = FileSystemStorage.get(zConf);
+ this.notebookDir = this.fs.makeQualified(new Path(zConf.getNotebookDir()));
+ LOGGER.info("Using folder {} to store notebook", notebookDir);
+ this.fs.tryMkDir(notebookDir);
- this.isSecurityEnabled = UserGroupInformation.isSecurityEnabled();
- if (isSecurityEnabled) {
- String keytab = zConf.getString(
- ZeppelinConfiguration.ConfVars.ZEPPELIN_SERVER_KERBEROS_KEYTAB);
- String principal = zConf.getString(
- ZeppelinConfiguration.ConfVars.ZEPPELIN_SERVER_KERBEROS_PRINCIPAL);
- if (StringUtils.isBlank(keytab) || StringUtils.isBlank(principal)) {
- throw new IOException("keytab and principal can not be empty, keytab: " + keytab
- + ", principal: " + principal);
- }
- UserGroupInformation.loginUserFromKeytab(principal, keytab);
- }
-
- try {
- this.fs = FileSystem.get(new URI(zConf.getNotebookDir()), new Configuration());
- LOGGER.info("Creating FileSystem: " + this.fs.getClass().getCanonicalName());
- this.notebookDir = fs.makeQualified(new Path(zConf.getNotebookDir()));
- LOGGER.info("Using folder {} to store notebook", notebookDir);
- } catch (URISyntaxException e) {
- throw new IOException(e);
- }
- if (!fs.exists(notebookDir)) {
- fs.mkdirs(notebookDir);
- LOGGER.info("Create notebook dir {} in hdfs", notebookDir.toString());
- }
- if (fs.isFile(notebookDir)) {
- throw new IOException("notebookDir {} is file instead of directory, please remove it or " +
- "specify another directory");
- }
}
@Override
public List list(AuthenticationInfo subject) throws IOException {
- return callHdfsOperation(new HdfsOperation>() {
- @Override
- public List call() throws IOException {
- List noteInfos = new ArrayList<>();
- for (FileStatus status : fs.globStatus(new Path(notebookDir, "*/note.json"))) {
- NoteInfo noteInfo = new NoteInfo(status.getPath().getParent().getName(), "", null);
- noteInfos.add(noteInfo);
- }
- return noteInfos;
- }
- });
+ List notePaths = fs.list(new Path(notebookDir, "*/note.json"));
+ List noteInfos = new ArrayList<>();
+ for (Path path : notePaths) {
+ NoteInfo noteInfo = new NoteInfo(path.getParent().getName(), "", null);
+ noteInfos.add(noteInfo);
+ }
+ return noteInfos;
}
@Override
public Note get(final String noteId, AuthenticationInfo subject) throws IOException {
- return callHdfsOperation(new HdfsOperation() {
- @Override
- public Note call() throws IOException {
- Path notePath = new Path(notebookDir.toString() + "/" + noteId + "/note.json");
- LOGGER.debug("Read note from file: " + notePath);
- ByteArrayOutputStream noteBytes = new ByteArrayOutputStream();
- IOUtils.copyBytes(fs.open(notePath), noteBytes, hadoopConf);
- return Note.fromJson(new String(noteBytes.toString(
- zConf.getString(ZeppelinConfiguration.ConfVars.ZEPPELIN_ENCODING))));
- }
- });
+ String content = this.fs.readFile(
+ new Path(notebookDir.toString() + "/" + noteId + "/note.json"));
+ return Note.fromJson(content);
}
@Override
public void save(final Note note, AuthenticationInfo subject) throws IOException {
- callHdfsOperation(new HdfsOperation() {
- @Override
- public Void call() throws IOException {
- Path notePath = new Path(notebookDir.toString() + "/" + note.getId() + "/note.json");
- Path tmpNotePath = new Path(notebookDir.toString() + "/" + note.getId() + "/.note.json");
- LOGGER.debug("Saving note to file: " + notePath);
- if (fs.exists(tmpNotePath)) {
- fs.delete(tmpNotePath, true);
- }
- InputStream in = new ByteArrayInputStream(note.toJson().getBytes(
- zConf.getString(ZeppelinConfiguration.ConfVars.ZEPPELIN_ENCODING)));
- IOUtils.copyBytes(in, fs.create(tmpNotePath), hadoopConf);
- fs.delete(notePath, true);
- fs.rename(tmpNotePath, notePath);
- return null;
- }
- });
+ this.fs.writeFile(note.toJson(),
+ new Path(notebookDir.toString() + "/" + note.getId() + "/note.json"),
+ true);
}
@Override
public void remove(final String noteId, AuthenticationInfo subject) throws IOException {
- callHdfsOperation(new HdfsOperation() {
- @Override
- public Void call() throws IOException {
- Path noteFolder = new Path(notebookDir.toString() + "/" + noteId);
- fs.delete(noteFolder, true);
- return null;
- }
- });
+ this.fs.delete(new Path(notebookDir.toString() + "/" + noteId));
}
@Override
@@ -182,26 +120,4 @@ public class FileSystemNotebookRepo implements NotebookRepo {
public void updateSettings(Map settings, AuthenticationInfo subject) {
LOGGER.warn("updateSettings is not implemented for HdfsNotebookRepo");
}
-
- private interface HdfsOperation {
- T call() throws IOException;
- }
-
- public synchronized T callHdfsOperation(final HdfsOperation func) throws IOException {
- if (isSecurityEnabled) {
- UserGroupInformation.getLoginUser().reloginFromKeytab();
- try {
- return UserGroupInformation.getCurrentUser().doAs(new PrivilegedExceptionAction() {
- @Override
- public T run() throws Exception {
- return func.call();
- }
- });
- } catch (InterruptedException e) {
- throw new IOException(e);
- }
- } else {
- return func.call();
- }
- }
}
diff --git a/zeppelin-zengine/src/main/java/org/apache/zeppelin/util/ReflectionUtils.java b/zeppelin-zengine/src/main/java/org/apache/zeppelin/util/ReflectionUtils.java
new file mode 100644
index 0000000000..ca09992a7d
--- /dev/null
+++ b/zeppelin-zengine/src/main/java/org/apache/zeppelin/util/ReflectionUtils.java
@@ -0,0 +1,99 @@
+/*
+ * 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.util;
+
+import java.io.IOException;
+import java.lang.reflect.Constructor;
+import java.lang.reflect.InvocationTargetException;
+
+
+/**
+ * Utility class for creating instances via java reflection.
+ *
+ */
+public class ReflectionUtils {
+
+ public static Class getClazz(String className) throws IOException {
+ Class clazz = null;
+ try {
+ clazz = Class.forName(className, true, Thread.currentThread().getContextClassLoader());
+ } catch (ClassNotFoundException e) {
+ throw new IOException("Unable to load class: " + className, e);
+ }
+
+ return clazz;
+ }
+
+ private static T getNewInstance(Class clazz) throws IOException {
+ T instance;
+ try {
+ instance = clazz.newInstance();
+ } catch (InstantiationException e) {
+ throw new IOException(
+ "Unable to instantiate class with 0 arguments: " + clazz.getName(), e);
+ } catch (IllegalAccessException e) {
+ throw new IOException(
+ "Unable to instantiate class with 0 arguments: " + clazz.getName(), e);
+ }
+ return instance;
+ }
+
+ private static T getNewInstance(Class clazz,
+ Class[] parameterTypes,
+ Object[] parameters)
+ throws IOException {
+ T instance;
+ try {
+ Constructor constructor = clazz.getConstructor(parameterTypes);
+ instance = constructor.newInstance(parameters);
+ } catch (InstantiationException e) {
+ throw new IOException(
+ "Unable to instantiate class with " + parameters.length + " arguments: " +
+ clazz.getName(), e);
+ } catch (IllegalAccessException e) {
+ throw new IOException(
+ "Unable to instantiate class with " + parameters.length + " arguments: " +
+ clazz.getName(), e);
+ } catch (NoSuchMethodException e) {
+ throw new IOException(
+ "Unable to instantiate class with " + parameters.length + " arguments: " +
+ clazz.getName(), e);
+ } catch (InvocationTargetException e) {
+ throw new IOException(
+ "Unable to instantiate class with " + parameters.length + " arguments: " +
+ clazz.getName(), e);
+ }
+ return instance;
+ }
+
+ public static T createClazzInstance(String className) throws IOException {
+ Class clazz = getClazz(className);
+ @SuppressWarnings("unchecked")
+ T instance = (T) getNewInstance(clazz);
+ return instance;
+ }
+
+ public static T createClazzInstance(String className,
+ Class[] parameterTypes,
+ Object[] parameters) throws IOException {
+ Class clazz = getClazz(className);
+ T instance = (T) getNewInstance(clazz, parameterTypes, parameters);
+ return instance;
+ }
+
+
+}
diff --git a/zeppelin-zengine/src/test/java/org/apache/zeppelin/interpreter/AbstractInterpreterTest.java b/zeppelin-zengine/src/test/java/org/apache/zeppelin/interpreter/AbstractInterpreterTest.java
index 9df402d35e..16c8c1d8ce 100644
--- a/zeppelin-zengine/src/test/java/org/apache/zeppelin/interpreter/AbstractInterpreterTest.java
+++ b/zeppelin-zengine/src/test/java/org/apache/zeppelin/interpreter/AbstractInterpreterTest.java
@@ -33,7 +33,7 @@ public abstract class AbstractInterpreterTest {
protected File interpreterDir;
protected File confDir;
protected File notebookDir;
- protected ZeppelinConfiguration conf = new ZeppelinConfiguration();
+ protected ZeppelinConfiguration conf;
@Before
public void setUp() throws Exception {
diff --git a/zeppelin-zengine/src/test/java/org/apache/zeppelin/interpreter/ConfInterpreterTest.java b/zeppelin-zengine/src/test/java/org/apache/zeppelin/interpreter/ConfInterpreterTest.java
new file mode 100644
index 0000000000..4d74c7cbfb
--- /dev/null
+++ b/zeppelin-zengine/src/test/java/org/apache/zeppelin/interpreter/ConfInterpreterTest.java
@@ -0,0 +1,102 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.zeppelin.interpreter;
+
+import com.sun.net.httpserver.Authenticator;
+import org.apache.zeppelin.display.GUI;
+import org.apache.zeppelin.interpreter.remote.RemoteInterpreter;
+import org.apache.zeppelin.user.AuthenticationInfo;
+import org.junit.Test;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertTrue;
+
+public class ConfInterpreterTest extends AbstractInterpreterTest {
+
+ @Test
+ public void testCorrectConf() throws IOException, InterpreterException {
+ interpreterSettingManager.setInterpreterBinding("user1", "note1", interpreterSettingManager.getSettingIds());
+ assertTrue(interpreterFactory.getInterpreter("user1", "note1", "test.conf") instanceof ConfInterpreter);
+ ConfInterpreter confInterpreter = (ConfInterpreter) interpreterFactory.getInterpreter("user1", "note1", "test.conf");
+
+ InterpreterContext context = new InterpreterContext("noteId", "paragraphId", "repl",
+ "title", "text", AuthenticationInfo.ANONYMOUS, new HashMap(), new GUI(), new GUI(),
+ null, null, new ArrayList(), null);
+ InterpreterResult result = confInterpreter.interpret("property_1\tnew_value\nnew_property\tdummy_value", context);
+ assertEquals(InterpreterResult.Code.SUCCESS, result.code);
+
+ assertTrue(interpreterFactory.getInterpreter("user1", "note1", "test") instanceof RemoteInterpreter);
+ RemoteInterpreter remoteInterpreter = (RemoteInterpreter) interpreterFactory.getInterpreter("user1", "note1", "test");
+ remoteInterpreter.interpret("hello world", context);
+ assertEquals(7, remoteInterpreter.getProperties().size());
+ assertEquals("new_value", remoteInterpreter.getProperty("property_1"));
+ assertEquals("dummy_value", remoteInterpreter.getProperty("new_property"));
+ assertEquals("value_3", remoteInterpreter.getProperty("property_3"));
+
+ // rerun the paragraph with the same properties would result in SUCCESS
+ result = confInterpreter.interpret("property_1\tnew_value\nnew_property\tdummy_value", context);
+ assertEquals(InterpreterResult.Code.SUCCESS, result.code);
+
+ // run the paragraph with the same properties would result in ERROR
+ result = confInterpreter.interpret("property_1\tnew_value_2\nnew_property\tdummy_value", context);
+ assertEquals(InterpreterResult.Code.ERROR, result.code);
+ }
+
+ @Test
+ public void testEmptyConf() throws IOException, InterpreterException {
+ interpreterSettingManager.setInterpreterBinding("user1", "note1", interpreterSettingManager.getSettingIds());
+ assertTrue(interpreterFactory.getInterpreter("user1", "note1", "test.conf") instanceof ConfInterpreter);
+ ConfInterpreter confInterpreter = (ConfInterpreter) interpreterFactory.getInterpreter("user1", "note1", "test.conf");
+
+ InterpreterContext context = new InterpreterContext("noteId", "paragraphId", "repl",
+ "title", "text", AuthenticationInfo.ANONYMOUS, new HashMap(), new GUI(), new GUI(),
+ null, null, new ArrayList(), null);
+ InterpreterResult result = confInterpreter.interpret("", context);
+ assertEquals(InterpreterResult.Code.SUCCESS, result.code);
+
+ assertTrue(interpreterFactory.getInterpreter("user1", "note1", "test") instanceof RemoteInterpreter);
+ RemoteInterpreter remoteInterpreter = (RemoteInterpreter) interpreterFactory.getInterpreter("user1", "note1", "test");
+ assertEquals(6, remoteInterpreter.getProperties().size());
+ assertEquals("value_1", remoteInterpreter.getProperty("property_1"));
+ assertEquals("value_3", remoteInterpreter.getProperty("property_3"));
+ }
+
+
+ @Test
+ public void testRunningAfterOtherInterpreter() throws IOException, InterpreterException {
+ interpreterSettingManager.setInterpreterBinding("user1", "note1", interpreterSettingManager.getSettingIds());
+ assertTrue(interpreterFactory.getInterpreter("user1", "note1", "test.conf") instanceof ConfInterpreter);
+ ConfInterpreter confInterpreter = (ConfInterpreter) interpreterFactory.getInterpreter("user1", "note1", "test.conf");
+
+ InterpreterContext context = new InterpreterContext("noteId", "paragraphId", "repl",
+ "title", "text", AuthenticationInfo.ANONYMOUS, new HashMap(), new GUI(), new GUI(),
+ null, null, new ArrayList(), null);
+ RemoteInterpreter remoteInterpreter = (RemoteInterpreter) interpreterFactory.getInterpreter("user1", "note1", "test");
+ InterpreterResult result = remoteInterpreter.interpret("hello world", context);
+ assertEquals(InterpreterResult.Code.SUCCESS, result.code);
+
+ result = confInterpreter.interpret("property_1\tnew_value\nnew_property\tdummy_value", context);
+ assertEquals(InterpreterResult.Code.ERROR, result.code);
+ }
+
+}
diff --git a/zeppelin-zengine/src/test/java/org/apache/zeppelin/interpreter/ManagedInterpreterGroupTest.java b/zeppelin-zengine/src/test/java/org/apache/zeppelin/interpreter/ManagedInterpreterGroupTest.java
index 74bd201085..aa7374991b 100644
--- a/zeppelin-zengine/src/test/java/org/apache/zeppelin/interpreter/ManagedInterpreterGroupTest.java
+++ b/zeppelin-zengine/src/test/java/org/apache/zeppelin/interpreter/ManagedInterpreterGroupTest.java
@@ -62,7 +62,7 @@ public class ManagedInterpreterGroupTest {
// create session_1
List interpreters = interpreterGroup.getOrCreateSession("user1", "session_1");
- assertEquals(2, interpreters.size());
+ assertEquals(3, interpreters.size());
assertEquals(EchoInterpreter.class.getName(), interpreters.get(0).getClassName());
assertEquals(DoubleEchoInterpreter.class.getName(), interpreters.get(1).getClassName());
assertEquals(1, interpreterGroup.getSessionNum());
@@ -73,7 +73,7 @@ public class ManagedInterpreterGroupTest {
// create session_2
List interpreters2 = interpreterGroup.getOrCreateSession("user1", "session_2");
- assertEquals(2, interpreters2.size());
+ assertEquals(3, interpreters2.size());
assertEquals(EchoInterpreter.class.getName(), interpreters2.get(0).getClassName());
assertEquals(DoubleEchoInterpreter.class.getName(), interpreters2.get(1).getClassName());
assertEquals(2, interpreterGroup.getSessionNum());
diff --git a/zeppelin-zengine/src/test/java/org/apache/zeppelin/interpreter/launcher/ShellScriptLauncherTest.java b/zeppelin-zengine/src/test/java/org/apache/zeppelin/interpreter/launcher/ShellScriptLauncherTest.java
index 0c7f4baacf..f7988e3570 100644
--- a/zeppelin-zengine/src/test/java/org/apache/zeppelin/interpreter/launcher/ShellScriptLauncherTest.java
+++ b/zeppelin-zengine/src/test/java/org/apache/zeppelin/interpreter/launcher/ShellScriptLauncherTest.java
@@ -22,6 +22,7 @@ import org.apache.zeppelin.interpreter.InterpreterOption;
import org.apache.zeppelin.interpreter.remote.RemoteInterpreterManagedProcess;
import org.junit.Test;
+import java.io.IOException;
import java.util.Properties;
import static org.junit.Assert.assertEquals;
@@ -30,14 +31,14 @@ import static org.junit.Assert.assertTrue;
public class ShellScriptLauncherTest {
@Test
- public void testLauncher() {
+ public void testLauncher() throws IOException {
ZeppelinConfiguration zConf = new ZeppelinConfiguration();
- ShellScriptLauncher launcher = new ShellScriptLauncher(zConf);
+ ShellScriptLauncher launcher = new ShellScriptLauncher(zConf, null);
Properties properties = new Properties();
properties.setProperty("ENV_1", "VALUE_1");
properties.setProperty("property_1", "value_1");
InterpreterOption option = new InterpreterOption();
- InterpreterLaunchContext context = new InterpreterLaunchContext(properties, option, null, "groupId", "groupName", "name");
+ InterpreterLaunchContext context = new InterpreterLaunchContext(properties, option, null, "intpGroupId", "groupId", "groupName", "name");
InterpreterClient client = launcher.launch(context);
assertTrue( client instanceof RemoteInterpreterManagedProcess);
RemoteInterpreterManagedProcess interpreterProcess = (RemoteInterpreterManagedProcess) client;
diff --git a/zeppelin-zengine/src/test/java/org/apache/zeppelin/interpreter/launcher/SparkInterpreterLauncherTest.java b/zeppelin-zengine/src/test/java/org/apache/zeppelin/interpreter/launcher/SparkInterpreterLauncherTest.java
index b788ebdeee..3d7e251b07 100644
--- a/zeppelin-zengine/src/test/java/org/apache/zeppelin/interpreter/launcher/SparkInterpreterLauncherTest.java
+++ b/zeppelin-zengine/src/test/java/org/apache/zeppelin/interpreter/launcher/SparkInterpreterLauncherTest.java
@@ -22,6 +22,7 @@ import org.apache.zeppelin.interpreter.InterpreterOption;
import org.apache.zeppelin.interpreter.remote.RemoteInterpreterManagedProcess;
import org.junit.Test;
+import java.io.IOException;
import java.util.Properties;
import static org.junit.Assert.assertEquals;
@@ -30,9 +31,9 @@ import static org.junit.Assert.assertTrue;
public class SparkInterpreterLauncherTest {
@Test
- public void testLocalMode() {
+ public void testLocalMode() throws IOException {
ZeppelinConfiguration zConf = new ZeppelinConfiguration();
- SparkInterpreterLauncher launcher = new SparkInterpreterLauncher(zConf);
+ SparkInterpreterLauncher launcher = new SparkInterpreterLauncher(zConf, null);
Properties properties = new Properties();
properties.setProperty("SPARK_HOME", "/user/spark");
properties.setProperty("property_1", "value_1");
@@ -41,7 +42,7 @@ public class SparkInterpreterLauncherTest {
properties.setProperty("spark.jars", "jar_1");
InterpreterOption option = new InterpreterOption();
- InterpreterLaunchContext context = new InterpreterLaunchContext(properties, option, null, "groupId", "spark", "spark");
+ InterpreterLaunchContext context = new InterpreterLaunchContext(properties, option, null, "intpGroupId", "groupId", "spark", "spark");
InterpreterClient client = launcher.launch(context);
assertTrue( client instanceof RemoteInterpreterManagedProcess);
RemoteInterpreterManagedProcess interpreterProcess = (RemoteInterpreterManagedProcess) client;
@@ -55,9 +56,9 @@ public class SparkInterpreterLauncherTest {
}
@Test
- public void testYarnClientMode_1() {
+ public void testYarnClientMode_1() throws IOException {
ZeppelinConfiguration zConf = new ZeppelinConfiguration();
- SparkInterpreterLauncher launcher = new SparkInterpreterLauncher(zConf);
+ SparkInterpreterLauncher launcher = new SparkInterpreterLauncher(zConf, null);
Properties properties = new Properties();
properties.setProperty("SPARK_HOME", "/user/spark");
properties.setProperty("property_1", "value_1");
@@ -66,7 +67,7 @@ public class SparkInterpreterLauncherTest {
properties.setProperty("spark.jars", "jar_1");
InterpreterOption option = new InterpreterOption();
- InterpreterLaunchContext context = new InterpreterLaunchContext(properties, option, null, "groupId", "spark", "spark");
+ InterpreterLaunchContext context = new InterpreterLaunchContext(properties, option, null, "intpGroupId", "groupId", "spark", "spark");
InterpreterClient client = launcher.launch(context);
assertTrue( client instanceof RemoteInterpreterManagedProcess);
RemoteInterpreterManagedProcess interpreterProcess = (RemoteInterpreterManagedProcess) client;
@@ -80,9 +81,9 @@ public class SparkInterpreterLauncherTest {
}
@Test
- public void testYarnClientMode_2() {
+ public void testYarnClientMode_2() throws IOException {
ZeppelinConfiguration zConf = new ZeppelinConfiguration();
- SparkInterpreterLauncher launcher = new SparkInterpreterLauncher(zConf);
+ SparkInterpreterLauncher launcher = new SparkInterpreterLauncher(zConf, null);
Properties properties = new Properties();
properties.setProperty("SPARK_HOME", "/user/spark");
properties.setProperty("property_1", "value_1");
@@ -92,7 +93,7 @@ public class SparkInterpreterLauncherTest {
properties.setProperty("spark.jars", "jar_1");
InterpreterOption option = new InterpreterOption();
- InterpreterLaunchContext context = new InterpreterLaunchContext(properties, option, null, "groupId", "spark", "spark");
+ InterpreterLaunchContext context = new InterpreterLaunchContext(properties, option, null, "intpGroupId", "groupId", "spark", "spark");
InterpreterClient client = launcher.launch(context);
assertTrue( client instanceof RemoteInterpreterManagedProcess);
RemoteInterpreterManagedProcess interpreterProcess = (RemoteInterpreterManagedProcess) client;
@@ -106,9 +107,9 @@ public class SparkInterpreterLauncherTest {
}
@Test
- public void testYarnClusterMode_1() {
+ public void testYarnClusterMode_1() throws IOException {
ZeppelinConfiguration zConf = new ZeppelinConfiguration();
- SparkInterpreterLauncher launcher = new SparkInterpreterLauncher(zConf);
+ SparkInterpreterLauncher launcher = new SparkInterpreterLauncher(zConf, null);
Properties properties = new Properties();
properties.setProperty("SPARK_HOME", "/user/spark");
properties.setProperty("property_1", "value_1");
@@ -117,7 +118,7 @@ public class SparkInterpreterLauncherTest {
properties.setProperty("spark.jars", "jar_1");
InterpreterOption option = new InterpreterOption();
- InterpreterLaunchContext context = new InterpreterLaunchContext(properties, option, null, "groupId", "spark", "spark");
+ InterpreterLaunchContext context = new InterpreterLaunchContext(properties, option, null, "intpGroupId", "groupId", "spark", "spark");
InterpreterClient client = launcher.launch(context);
assertTrue( client instanceof RemoteInterpreterManagedProcess);
RemoteInterpreterManagedProcess interpreterProcess = (RemoteInterpreterManagedProcess) client;
@@ -132,9 +133,9 @@ public class SparkInterpreterLauncherTest {
}
@Test
- public void testYarnClusterMode_2() {
+ public void testYarnClusterMode_2() throws IOException {
ZeppelinConfiguration zConf = new ZeppelinConfiguration();
- SparkInterpreterLauncher launcher = new SparkInterpreterLauncher(zConf);
+ SparkInterpreterLauncher launcher = new SparkInterpreterLauncher(zConf, null);
Properties properties = new Properties();
properties.setProperty("SPARK_HOME", "/user/spark");
properties.setProperty("property_1", "value_1");
@@ -144,7 +145,7 @@ public class SparkInterpreterLauncherTest {
properties.setProperty("spark.jars", "jar_1");
InterpreterOption option = new InterpreterOption();
- InterpreterLaunchContext context = new InterpreterLaunchContext(properties, option, null, "groupId", "spark", "spark");
+ InterpreterLaunchContext context = new InterpreterLaunchContext(properties, option, null, "intpGroupId", "groupId", "spark", "spark");
InterpreterClient client = launcher.launch(context);
assertTrue( client instanceof RemoteInterpreterManagedProcess);
RemoteInterpreterManagedProcess interpreterProcess = (RemoteInterpreterManagedProcess) client;
diff --git a/zeppelin-zengine/src/test/java/org/apache/zeppelin/interpreter/recovery/FileSystemRecoveryStorageTest.java b/zeppelin-zengine/src/test/java/org/apache/zeppelin/interpreter/recovery/FileSystemRecoveryStorageTest.java
new file mode 100644
index 0000000000..cf1899c13e
--- /dev/null
+++ b/zeppelin-zengine/src/test/java/org/apache/zeppelin/interpreter/recovery/FileSystemRecoveryStorageTest.java
@@ -0,0 +1,92 @@
+package org.apache.zeppelin.interpreter.recovery;
+
+import com.google.common.io.Files;
+import org.apache.commons.io.FileUtils;
+import org.apache.zeppelin.conf.ZeppelinConfiguration;
+import org.apache.zeppelin.display.GUI;
+import org.apache.zeppelin.interpreter.AbstractInterpreterTest;
+import org.apache.zeppelin.interpreter.Interpreter;
+import org.apache.zeppelin.interpreter.InterpreterContext;
+import org.apache.zeppelin.interpreter.InterpreterContextRunner;
+import org.apache.zeppelin.interpreter.InterpreterException;
+import org.apache.zeppelin.interpreter.InterpreterOption;
+import org.apache.zeppelin.interpreter.InterpreterSetting;
+import org.apache.zeppelin.interpreter.remote.RemoteInterpreter;
+import org.apache.zeppelin.user.AuthenticationInfo;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+
+import static org.junit.Assert.assertEquals;
+
+public class FileSystemRecoveryStorageTest extends AbstractInterpreterTest {
+
+ private File recoveryDir = null;
+
+ @Before
+ public void setUp() throws Exception {
+ System.setProperty(ZeppelinConfiguration.ConfVars.ZEPPELIN_RECOVERY_STORAGE_CLASS.getVarName(),
+ FileSystemRecoveryStorage.class.getName());
+ recoveryDir = Files.createTempDir();
+ System.setProperty(ZeppelinConfiguration.ConfVars.ZEPPELIN_RECOVERY_DIR.getVarName(), recoveryDir.getAbsolutePath());
+ super.setUp();
+ }
+
+ @Override
+ public void tearDown() throws Exception {
+ super.tearDown();
+ FileUtils.deleteDirectory(recoveryDir);
+ }
+
+ @Test
+ public void testSingleInterpreterProcess() throws InterpreterException, IOException {
+ InterpreterSetting interpreterSetting = interpreterSettingManager.getByName("test");
+ interpreterSetting.getOption().setPerUser(InterpreterOption.SHARED);
+
+ Interpreter interpreter1 = interpreterSetting.getDefaultInterpreter("user1", "note1");
+ RemoteInterpreter remoteInterpreter1 = (RemoteInterpreter) interpreter1;
+ InterpreterContext context1 = new InterpreterContext("noteId", "paragraphId", "repl",
+ "title", "text", AuthenticationInfo.ANONYMOUS, new HashMap(), new GUI(),
+ new GUI(), null, null, new ArrayList(), null);
+ remoteInterpreter1.interpret("hello", context1);
+
+ assertEquals(1, interpreterSettingManager.getRecoveryStorage().restore().size());
+
+ interpreterSetting.close();
+ assertEquals(0, interpreterSettingManager.getRecoveryStorage().restore().size());
+ }
+
+ @Test
+ public void testMultipleInterpreterProcess() throws InterpreterException, IOException {
+ InterpreterSetting interpreterSetting = interpreterSettingManager.getByName("test");
+ interpreterSetting.getOption().setPerUser(InterpreterOption.ISOLATED);
+
+ Interpreter interpreter1 = interpreterSetting.getDefaultInterpreter("user1", "note1");
+ RemoteInterpreter remoteInterpreter1 = (RemoteInterpreter) interpreter1;
+ InterpreterContext context1 = new InterpreterContext("noteId", "paragraphId", "repl",
+ "title", "text", AuthenticationInfo.ANONYMOUS, new HashMap(), new GUI(),
+ new GUI(), null, null, new ArrayList(), null);
+ remoteInterpreter1.interpret("hello", context1);
+ assertEquals(1, interpreterSettingManager.getRecoveryStorage().restore().size());
+
+ Interpreter interpreter2 = interpreterSetting.getDefaultInterpreter("user2", "note2");
+ RemoteInterpreter remoteInterpreter2 = (RemoteInterpreter) interpreter2;
+ InterpreterContext context2 = new InterpreterContext("noteId", "paragraphId", "repl",
+ "title", "text", AuthenticationInfo.ANONYMOUS, new HashMap(), new GUI(),
+ new GUI(), null, null, new ArrayList(), null);
+ remoteInterpreter2.interpret("hello", context2);
+
+ assertEquals(2, interpreterSettingManager.getRecoveryStorage().restore().size());
+
+ interpreterSettingManager.restart(interpreterSetting.getId(), "note1", "user1");
+ assertEquals(1, interpreterSettingManager.getRecoveryStorage().restore().size());
+
+ interpreterSetting.close();
+ assertEquals(0, interpreterSettingManager.getRecoveryStorage().restore().size());
+ }
+
+}
diff --git a/zeppelin-zengine/src/test/java/org/apache/zeppelin/notebook/ParagraphTest.java b/zeppelin-zengine/src/test/java/org/apache/zeppelin/notebook/ParagraphTest.java
index 9e9ce27ea9..e46b7393ee 100644
--- a/zeppelin-zengine/src/test/java/org/apache/zeppelin/notebook/ParagraphTest.java
+++ b/zeppelin-zengine/src/test/java/org/apache/zeppelin/notebook/ParagraphTest.java
@@ -271,7 +271,6 @@ public class ParagraphTest extends AbstractInterpreterTest {
@Test
public void testCursorPosition() {
Paragraph paragraph = spy(new Paragraph());
- doReturn(null).when(paragraph).getIntpText();
// left = buffer, middle = cursor position into source code, right = cursor position after parse
List> dataSet = Arrays.asList(
Triple.of("%jdbc schema.", 13, 7),
@@ -294,7 +293,8 @@ public class ParagraphTest extends AbstractInterpreterTest {
);
for (Triple data : dataSet) {
- Integer actual = paragraph.calculateCursorPosition(data.getLeft(), data.getLeft().trim(), data.getMiddle());
+ paragraph.setText(data.getLeft());
+ Integer actual = paragraph.calculateCursorPosition(data.getLeft(), data.getMiddle());
assertEquals(data.getRight(), actual);
}
}
+
+
+## Interpreter Process Recovery
+
+Before 0.8.0, shutting down Zeppelin also mean to shutdown all the running interpreter processes. Usually admin will shutdown Zeppelin server for maintenance or upgrade, but don't want to shut down the running interpreter processes.
+In such cases, interpreter process recovery is necessary. Starting from 0.8.0, user can enable interpreter process recovering via setting `zeppelin.recovery.storage.class` as
+`org.apache.zeppelin.interpreter.recovery.FileSystemRecoveryStorage` or other implementations if available in future, by default it is `org.apache.zeppelin.interpreter.recovery.NullRecoveryStorage`
+ which means recovery is not enabled. Enable recover means shutting down Zeppelin would not terminating interpreter process,
+and when Zeppelin is restarted, it would try to reconnect to the existing running interpreter processes. If you want to kill all the interpreter processes after terminating Zeppelin even when recovery is enabled, you can run `bin/stop-interpreter.sh`
diff --git a/kylin/src/main/java/org/apache/zeppelin/kylin/KylinErrorResponse.java b/kylin/src/main/java/org/apache/zeppelin/kylin/KylinErrorResponse.java
new file mode 100644
index 0000000000..00439e8c62
--- /dev/null
+++ b/kylin/src/main/java/org/apache/zeppelin/kylin/KylinErrorResponse.java
@@ -0,0 +1,63 @@
+/*
+ * 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.kylin;
+
+import com.google.gson.Gson;
+import com.google.gson.JsonSyntaxException;
+import org.apache.zeppelin.common.JsonSerializable;
+
+/**
+ * class for Kylin Error Response.
+ */
+class KylinErrorResponse implements JsonSerializable {
+ private static final Gson gson = new Gson();
+
+ private String stacktrace;
+ private String exception;
+ private String url;
+ private String code;
+ private Object data;
+ private String msg;
+
+ public KylinErrorResponse(String stacktrace, String exception, String url,
+ String code, Object data, String msg) {
+ this.stacktrace = stacktrace;
+ this.exception = exception;
+ this.url = url;
+ this.code = code;
+ this.data = data;
+ this.msg = msg;
+ }
+
+ public String getException() {
+ return exception;
+ }
+
+ public String toJson() {
+ return gson.toJson(this);
+ }
+
+ public static KylinErrorResponse fromJson(String json) {
+ try {
+ return gson.fromJson(json, KylinErrorResponse.class);
+ } catch (JsonSyntaxException ex) {
+ return null;
+ }
+ }
+
+}
diff --git a/kylin/src/main/java/org/apache/zeppelin/kylin/KylinInterpreter.java b/kylin/src/main/java/org/apache/zeppelin/kylin/KylinInterpreter.java
index 6b68d288e4..c7cd689a74 100755
--- a/kylin/src/main/java/org/apache/zeppelin/kylin/KylinInterpreter.java
+++ b/kylin/src/main/java/org/apache/zeppelin/kylin/KylinInterpreter.java
@@ -18,6 +18,7 @@
package org.apache.zeppelin.kylin;
import org.apache.commons.codec.binary.Base64;
+import org.apache.commons.io.IOUtils;
import org.apache.http.HttpResponse;
import org.apache.http.client.HttpClient;
import org.apache.http.client.methods.HttpPost;
@@ -30,9 +31,7 @@ import org.apache.zeppelin.interpreter.thrift.InterpreterCompletion;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import java.io.BufferedReader;
import java.io.IOException;
-import java.io.InputStreamReader;
import java.util.List;
import java.util.Properties;
import java.util.regex.Matcher;
@@ -166,28 +165,42 @@ public class KylinInterpreter extends Interpreter {
}
private InterpreterResult executeQuery(String sql) throws IOException {
-
HttpResponse response = prepareRequest(sql);
+ String result;
- if (response.getStatusLine().getStatusCode() != 200) {
- logger.error("failed to execute query: " + response.getEntity().getContent().toString());
- return new InterpreterResult(InterpreterResult.Code.ERROR,
- "Failed : HTTP error code " + response.getStatusLine().getStatusCode());
+ try {
+ int code = response.getStatusLine().getStatusCode();
+ result = IOUtils.toString(response.getEntity().getContent(), "UTF-8");
+
+ if (code != 200) {
+ StringBuilder errorMessage = new StringBuilder("Failed : HTTP error code " + code + " .");
+ logger.error("Failed to execute query: " + result);
+
+ KylinErrorResponse kylinErrorResponse = KylinErrorResponse.fromJson(result);
+ if (kylinErrorResponse == null) {
+ logger.error("Cannot get json from string: " + result);
+ // when code is 401, the response is html, not json
+ if (code == 401) {
+ errorMessage.append(" Error message: Unauthorized. This request requires "
+ + "HTTP authentication. Please make sure your have set your credentials "
+ + "correctly.");
+ } else {
+ errorMessage.append(" Error message: " + result + " .");
+ }
+ } else {
+ String exception = kylinErrorResponse.getException();
+ logger.error("The exception is " + exception);
+ errorMessage.append(" Error message: " + exception + " .");
+ }
+
+ return new InterpreterResult(InterpreterResult.Code.ERROR, errorMessage.toString());
+ }
+ } catch (NullPointerException | IOException e) {
+ throw new IOException(e);
}
- BufferedReader br = new BufferedReader(
- new InputStreamReader((response.getEntity().getContent())));
- StringBuilder sb = new StringBuilder();
-
- String output;
- logger.info("Output from Server .... \n");
- while ((output = br.readLine()) != null) {
- logger.info(output);
- sb.append(output).append('\n');
- }
- InterpreterResult rett = new InterpreterResult(InterpreterResult.Code.SUCCESS,
- formatResult(sb.toString()));
- return rett;
+ return new InterpreterResult(InterpreterResult.Code.SUCCESS,
+ formatResult(result));
}
String formatResult(String msg) {
@@ -205,16 +218,18 @@ public class KylinInterpreter extends Interpreter {
table = mr.group(1);
}
- String[] row = table.split("],\\[");
- for (int i = 0; i < row.length; i++) {
- String[] col = row[i].split(",(?=(?:[^\"]*\"[^\"]*\")*[^\"]*$)", -1);
- for (int j = 0; j < col.length; j++) {
- if (col[j] != null) {
- col[j] = col[j].replaceAll("^\"|\"$", "");
+ if (table != null && !table.isEmpty()) {
+ String[] row = table.split("],\\[");
+ for (int i = 0; i < row.length; i++) {
+ String[] col = row[i].split(",(?=(?:[^\"]*\"[^\"]*\")*[^\"]*$)", -1);
+ for (int j = 0; j < col.length; j++) {
+ if (col[j] != null) {
+ col[j] = col[j].replaceAll("^\"|\"$", "");
+ }
+ res.append(col[j] + " \t");
}
- res.append(col[j] + " \t");
+ res.append(" \n");
}
- res.append(" \n");
}
return res.toString();
}
diff --git a/kylin/src/test/java/org/apache/zeppelin/kylin/KylinInterpreterTest.java b/kylin/src/test/java/org/apache/zeppelin/kylin/KylinInterpreterTest.java
index 4471a07689..35f0f3c2eb 100755
--- a/kylin/src/test/java/org/apache/zeppelin/kylin/KylinInterpreterTest.java
+++ b/kylin/src/test/java/org/apache/zeppelin/kylin/KylinInterpreterTest.java
@@ -108,6 +108,30 @@ public class KylinInterpreterTest {
Assert.assertEquals(expected, actual);
}
+ @Test
+ public void testParseEmptyResult() {
+ String msg = "{\"columnMetas\":[{\"isNullable\":1,\"displaySize\":256,\"label\":\"COUNTRY\",\"name\":\"COUNTRY\","
+ + "\"schemaName\":\"DEFAULT\",\"catelogName\":null,\"tableName\":\"SALES_TABLE\",\"precision\":256,"
+ + "\"scale\":0,\"columnType\":12,\"columnTypeName\":\"VARCHAR\",\"writable\":false,\"readOnly\":true,"
+ + "\"definitelyWritable\":false,\"autoIncrement\":false,\"caseSensitive\":true,\"searchable\":false,"
+ + "\"currency\":false,\"signed\":true},{\"isNullable\":1,\"displaySize\":256,\"label\":\"CURRENCY\","
+ + "\"name\":\"CURRENCY\",\"schemaName\":\"DEFAULT\",\"catelogName\":null,\"tableName\":\"SALES_TABLE\","
+ + "\"precision\":256,\"scale\":0,\"columnType\":12,\"columnTypeName\":\"VARCHAR\",\"writable\":false,"
+ + "\"readOnly\":true,\"definitelyWritable\":false,\"autoIncrement\":false,\"caseSensitive\":true,"
+ + "\"searchable\":false,\"currency\":false,\"signed\":true},{\"isNullable\":0,\"displaySize\":19,"
+ + "\"label\":\"COUNT__\",\"name\":\"COUNT__\",\"schemaName\":\"DEFAULT\",\"catelogName\":null,"
+ + "\"tableName\":\"SALES_TABLE\",\"precision\":19,\"scale\":0,\"columnType\":-5,\"columnTypeName\":"
+ + "\"BIGINT\",\"writable\":false,\"readOnly\":true,\"definitelyWritable\":false,\"autoIncrement\":false,"
+ + "\"caseSensitive\":true,\"searchable\":false,\"currency\":false,\"signed\":true}],\"results\":"
+ + "[]," + "\"cube\":\"Sample_Cube\",\"affectedRowCount\":0,\"isException\":false,\"exceptionMessage\":null,"
+ + "\"duration\":134,\"totalScanCount\":1,\"hitExceptionCache\":false,\"storageCacheUsed\":false,"
+ + "\"partial\":false}";
+ String expected="%table COUNTRY \tCURRENCY \tCOUNT__ \t \n";
+ KylinInterpreter t = new MockKylinInterpreter(getDefaultProperties());
+ String actual = t.formatResult(msg);
+ Assert.assertEquals(expected, actual);
+ }
+
private Properties getDefaultProperties() {
Properties prop = new Properties();
prop.put("kylin.api.username", "ADMIN");
diff --git a/spark/src/main/resources/interpreter-setting.json b/spark/src/main/resources/interpreter-setting.json
index 485f6950df..d656532eb0 100644
--- a/spark/src/main/resources/interpreter-setting.json
+++ b/spark/src/main/resources/interpreter-setting.json
@@ -61,7 +61,7 @@
"description": "Spark master uri. ex) spark://masterhost:7077",
"type": "string"
},
- "zeppelin.spark.unSupportedVersionCheck": {
+ "zeppelin.spark.enableSupportedVersionCheck": {
"envName": null,
"propertyName": "zeppelin.spark.enableSupportedVersionCheck",
"defaultValue": true,
diff --git a/testing/install_external_dependencies.sh b/testing/install_external_dependencies.sh
index daa670bc7d..e34296e3ab 100755
--- a/testing/install_external_dependencies.sh
+++ b/testing/install_external_dependencies.sh
@@ -44,6 +44,6 @@ if [[ -n "$PYTHON" ]] ; then
conda update -q conda
conda info -a
conda config --add channels conda-forge
- conda install -q matplotlib pandasql ipython jupyter_client ipykernel matplotlib bokeh=0.12.6
+ conda install -q matplotlib pandasql ipython=5.4.1 jupyter_client ipykernel matplotlib bokeh=0.12.6
pip install -q grpcio ggplot
fi
diff --git a/zeppelin-distribution/src/bin_license/LICENSE b/zeppelin-distribution/src/bin_license/LICENSE
index ab120f235f..37fbce1b1f 100644
--- a/zeppelin-distribution/src/bin_license/LICENSE
+++ b/zeppelin-distribution/src/bin_license/LICENSE
@@ -311,7 +311,7 @@ The text of each license is also included at licenses/LICENSE-[project]-[version
The following components are provided under the BSD-style License.
- (New BSD License) JGit (org.eclipse.jgit:org.eclipse.jgit:jar:4.1.1.201511131810-r - https://eclipse.org/jgit/)
+ (New BSD License) JGit (org.eclipse.jgit:org.eclipse.jgit:jar:4.5.4.201711221230-r - https://eclipse.org/jgit/)
(New BSD License) Kryo (com.esotericsoftware.kryo:kryo:3.0.3 - http://code.google.com/p/kryo/)
(New BSD License) MinLog (com.esotericsoftware.minlog:minlog:1.3 - http://code.google.com/p/minlog/)
(New BSD License) ReflectASM (com.esotericsoftware.reflectasm:reflectasm:1.07 - http://code.google.com/p/reflectasm/)
diff --git a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/conf/ZeppelinConfiguration.java b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/conf/ZeppelinConfiguration.java
index 508ed8398b..992e90f7ed 100644
--- a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/conf/ZeppelinConfiguration.java
+++ b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/conf/ZeppelinConfiguration.java
@@ -355,6 +355,19 @@ public class ZeppelinConfiguration extends XMLConfiguration {
return getString(ConfVars.ZEPPELIN_NOTEBOOK_DIR);
}
+ public String getRecoveryDir() {
+ return getRelativeDir(ConfVars.ZEPPELIN_RECOVERY_DIR);
+ }
+
+ public String getRecoveryStorageClass() {
+ return getString(ConfVars.ZEPPELIN_RECOVERY_STORAGE_CLASS);
+ }
+
+ public boolean isRecoveryEnabled() {
+ return !getString(ConfVars.ZEPPELIN_RECOVERY_STORAGE_CLASS).equals(
+ "org.apache.zeppelin.interpreter.recovery.NullRecoveryStorage");
+ }
+
public String getUser() {
return getString(ConfVars.ZEPPELIN_NOTEBOOK_S3_USER);
}
@@ -674,6 +687,10 @@ public class ZeppelinConfiguration extends XMLConfiguration {
ZEPPELIN_INTERPRETER_OUTPUT_LIMIT("zeppelin.interpreter.output.limit", 1024 * 100),
ZEPPELIN_ENCODING("zeppelin.encoding", "UTF-8"),
ZEPPELIN_NOTEBOOK_DIR("zeppelin.notebook.dir", "notebook"),
+ ZEPPELIN_RECOVERY_DIR("zeppelin.recovery.dir", "recovery"),
+ ZEPPELIN_RECOVERY_STORAGE_CLASS("zeppelin.recovery.storage.class",
+ "org.apache.zeppelin.interpreter.recovery.NullRecoveryStorage"),
+
// use specified notebook (id) as homescreen
ZEPPELIN_NOTEBOOK_HOMESCREEN("zeppelin.notebook.homescreen", null),
// whether homescreen notebook will be hidden from notebook list or not
diff --git a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/launcher/InterpreterClient.java b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/launcher/InterpreterClient.java
index b991079fec..813dad8688 100644
--- a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/launcher/InterpreterClient.java
+++ b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/launcher/InterpreterClient.java
@@ -19,8 +19,20 @@ package org.apache.zeppelin.interpreter.launcher;
/**
* Interface to InterpreterClient which is created by InterpreterLauncher. This is the component
- * that is used to for the communication fromzeppelin-server process to zeppelin interpreter process
+ * that is used to for the communication from zeppelin-server process to zeppelin interpreter
+ * process.
*/
public interface InterpreterClient {
+ String getInterpreterSettingName();
+
+ void start(String userName, Boolean isUserImpersonate);
+
+ void stop();
+
+ String getHost();
+
+ int getPort();
+
+ boolean isRunning();
}
diff --git a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/launcher/InterpreterLaunchContext.java b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/launcher/InterpreterLaunchContext.java
index 9e253555a9..6901e2c7a6 100644
--- a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/launcher/InterpreterLaunchContext.java
+++ b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/launcher/InterpreterLaunchContext.java
@@ -30,6 +30,7 @@ public class InterpreterLaunchContext {
private Properties properties;
private InterpreterOption option;
private InterpreterRunner runner;
+ private String interpreterGroupId;
private String interpreterSettingId;
private String interpreterSettingGroup;
private String interpreterSettingName;
@@ -37,12 +38,14 @@ public class InterpreterLaunchContext {
public InterpreterLaunchContext(Properties properties,
InterpreterOption option,
InterpreterRunner runner,
+ String interpreterGroupId,
String interpreterSettingId,
String interpreterSettingGroup,
String interpreterSettingName) {
this.properties = properties;
this.option = option;
this.runner = runner;
+ this.interpreterGroupId = interpreterGroupId;
this.interpreterSettingId = interpreterSettingId;
this.interpreterSettingGroup = interpreterSettingGroup;
this.interpreterSettingName = interpreterSettingName;
@@ -60,6 +63,10 @@ public class InterpreterLaunchContext {
return runner;
}
+ public String getInterpreterGroupId() {
+ return interpreterGroupId;
+ }
+
public String getInterpreterSettingId() {
return interpreterSettingId;
}
diff --git a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/launcher/InterpreterLauncher.java b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/launcher/InterpreterLauncher.java
index 5d0acf3515..1cee20e7a0 100644
--- a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/launcher/InterpreterLauncher.java
+++ b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/launcher/InterpreterLauncher.java
@@ -18,6 +18,7 @@
package org.apache.zeppelin.interpreter.launcher;
import org.apache.zeppelin.conf.ZeppelinConfiguration;
+import org.apache.zeppelin.interpreter.recovery.RecoveryStorage;
import java.io.IOException;
import java.util.Properties;
@@ -29,9 +30,11 @@ public abstract class InterpreterLauncher {
protected ZeppelinConfiguration zConf;
protected Properties properties;
+ protected RecoveryStorage recoveryStorage;
- public InterpreterLauncher(ZeppelinConfiguration zConf) {
+ public InterpreterLauncher(ZeppelinConfiguration zConf, RecoveryStorage recoveryStorage) {
this.zConf = zConf;
+ this.recoveryStorage = recoveryStorage;
}
public abstract InterpreterClient launch(InterpreterLaunchContext context) throws IOException;
diff --git a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/recovery/RecoveryStorage.java b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/recovery/RecoveryStorage.java
new file mode 100644
index 0000000000..8bbe8302fc
--- /dev/null
+++ b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/recovery/RecoveryStorage.java
@@ -0,0 +1,80 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.zeppelin.interpreter.recovery;
+
+import org.apache.zeppelin.conf.ZeppelinConfiguration;
+import org.apache.zeppelin.interpreter.launcher.InterpreterClient;
+
+import java.io.IOException;
+import java.util.Map;
+
+
+/**
+ * Interface for storing interpreter process recovery metadata.
+ *
+ */
+public abstract class RecoveryStorage {
+
+ protected ZeppelinConfiguration zConf;
+ protected Map