You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@zeppelin.apache.org by zj...@apache.org on 2020/09/18 02:13:59 UTC
[zeppelin] branch master updated: [ZEPPELIN-5041]. Move interpreter
lifecyclemanager from zeppelin server to interpreter process
This is an automated email from the ASF dual-hosted git repository.
zjffdu pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/zeppelin.git
The following commit(s) were added to refs/heads/master by this push:
new 55375ed [ZEPPELIN-5041]. Move interpreter lifecyclemanager from zeppelin server to interpreter process
55375ed is described below
commit 55375edf487c06eb28d34d8ec7e3704522b69d90
Author: Jeff Zhang <zj...@apache.org>
AuthorDate: Mon Sep 14 15:11:02 2020 +0800
[ZEPPELIN-5041]. Move interpreter lifecyclemanager from zeppelin server to interpreter process
### What is this PR for?
Currently, interpreter lifecycle manager run in zeppelin server side. One short coming is that interpreter process may still leak if zeppelin server exist unexpectedly. This PR move lifecyclemanager from zeppelin server to interpreter process, so that if user use TimeoutLifecycleManager, interpreter process will exit after a idle time even zeppelin server is existed.
### What type of PR is it?
[ Improvement]
### Todos
* [ ] - Task
### What is the Jira issue?
* https://issues.apache.org/jira/browse/ZEPPELIN-5041
### How should this be tested?
* CI pass & manually tested
### Screenshots (if appropriate)
### Questions:
* Does the licenses files need update? No
* Is there breaking changes for older versions? No
* Does this needs documentation? No
Author: Jeff Zhang <zj...@apache.org>
Closes #3910 from zjffdu/ZEPPELIN-5041 and squashes the following commits:
5473d24d7 [Jeff Zhang] [ZEPPELIN-5041]. Move interpreter lifecyclemanager from zeppelin server to interpreter process
---
.../apache/zeppelin/helium/ZeppelinDevServer.java | 2 +-
.../zeppelin/integration/FlinkIntegrationTest.java | 2 +-
.../zeppelin/integration/JdbcIntegrationTest.java | 2 +-
.../apache/zeppelin/integration/MiniZeppelin.java | 7 +-
.../zeppelin/integration/SparkIntegrationTest.java | 2 +-
.../YarnInterpreterLauncherIntegrationTest.java | 2 +-
.../zeppelin/conf/ZeppelinConfiguration.java | 28 +-
.../zeppelin/interpreter/LifecycleManager.java | 20 +-
.../lifecycle/NullLifecycleManager.java | 13 +-
.../lifecycle/TimeoutLifecycleManager.java | 82 +++
.../remote/RemoteInterpreterEventClient.java | 7 +
.../remote/RemoteInterpreterServer.java | 59 +-
.../interpreter/thrift/AngularObjectId.java | 2 +-
.../interpreter/thrift/AppOutputAppendEvent.java | 2 +-
.../interpreter/thrift/AppOutputUpdateEvent.java | 2 +-
.../interpreter/thrift/AppStatusUpdateEvent.java | 2 +-
.../interpreter/thrift/InterpreterCompletion.java | 2 +-
.../interpreter/thrift/OutputAppendEvent.java | 2 +-
.../interpreter/thrift/OutputUpdateAllEvent.java | 2 +-
.../interpreter/thrift/OutputUpdateEvent.java | 2 +-
.../zeppelin/interpreter/thrift/ParagraphInfo.java | 2 +-
.../zeppelin/interpreter/thrift/RegisterInfo.java | 2 +-
.../thrift/RemoteApplicationResult.java | 2 +-
.../thrift/RemoteInterpreterContext.java | 2 +-
.../interpreter/thrift/RemoteInterpreterEvent.java | 2 +-
.../thrift/RemoteInterpreterEventService.java | 766 ++++++++++++++++++++-
.../thrift/RemoteInterpreterEventType.java | 2 +-
.../thrift/RemoteInterpreterResult.java | 2 +-
.../thrift/RemoteInterpreterResultMessage.java | 2 +-
.../thrift/RemoteInterpreterService.java | 2 +-
.../interpreter/thrift/RunParagraphsEvent.java | 2 +-
.../interpreter/thrift/ServiceException.java | 2 +-
.../zeppelin/interpreter/thrift/WebUrlInfo.java | 2 +-
.../thrift/RemoteInterpreterEventService.thrift | 1 +
.../remote/RemoteInterpreterServerTest.java | 6 +-
.../zeppelin/interpreter/InterpreterSetting.java | 21 +-
.../interpreter/InterpreterSettingManager.java | 18 +-
.../interpreter/ManagedInterpreterGroup.java | 1 -
.../interpreter/RemoteInterpreterEventServer.java | 26 +-
.../lifecycle/TimeoutLifecycleManager.java | 74 --
.../interpreter/remote/RemoteInterpreter.java | 14 +-
.../interpreter/InterpreterSettingManagerTest.java | 2 -
.../interpreter/SessionConfInterpreterTest.java | 2 +-
.../lifecycle/TimeoutLifecycleManagerTest.java | 19 +-
44 files changed, 1025 insertions(+), 191 deletions(-)
diff --git a/helium-dev/src/main/java/org/apache/zeppelin/helium/ZeppelinDevServer.java b/helium-dev/src/main/java/org/apache/zeppelin/helium/ZeppelinDevServer.java
index 4c0e867..e682981 100644
--- a/helium-dev/src/main/java/org/apache/zeppelin/helium/ZeppelinDevServer.java
+++ b/helium-dev/src/main/java/org/apache/zeppelin/helium/ZeppelinDevServer.java
@@ -38,7 +38,7 @@ public class ZeppelinDevServer extends
private DevInterpreter interpreter = null;
private InterpreterOutput out;
- public ZeppelinDevServer(int port) throws TException, IOException {
+ public ZeppelinDevServer(int port) throws Exception {
super(null, port, null, ":");
}
diff --git a/zeppelin-interpreter-integration/src/test/java/org/apache/zeppelin/integration/FlinkIntegrationTest.java b/zeppelin-interpreter-integration/src/test/java/org/apache/zeppelin/integration/FlinkIntegrationTest.java
index 6452989..44f0f2c 100644
--- a/zeppelin-interpreter-integration/src/test/java/org/apache/zeppelin/integration/FlinkIntegrationTest.java
+++ b/zeppelin-interpreter-integration/src/test/java/org/apache/zeppelin/integration/FlinkIntegrationTest.java
@@ -76,7 +76,7 @@ public abstract class FlinkIntegrationTest {
hadoopCluster.start();
zeppelin = new MiniZeppelin();
- zeppelin.start();
+ zeppelin.start(FlinkIntegrationTest.class);
interpreterFactory = zeppelin.getInterpreterFactory();
interpreterSettingManager = zeppelin.getInterpreterSettingManager();
}
diff --git a/zeppelin-interpreter-integration/src/test/java/org/apache/zeppelin/integration/JdbcIntegrationTest.java b/zeppelin-interpreter-integration/src/test/java/org/apache/zeppelin/integration/JdbcIntegrationTest.java
index d9e7d35..81ad481 100644
--- a/zeppelin-interpreter-integration/src/test/java/org/apache/zeppelin/integration/JdbcIntegrationTest.java
+++ b/zeppelin-interpreter-integration/src/test/java/org/apache/zeppelin/integration/JdbcIntegrationTest.java
@@ -47,7 +47,7 @@ public class JdbcIntegrationTest {
@BeforeClass
public static void setUp() throws IOException {
zeppelin = new MiniZeppelin();
- zeppelin.start();
+ zeppelin.start(JdbcIntegrationTest.class);
interpreterFactory = zeppelin.getInterpreterFactory();
interpreterSettingManager = zeppelin.getInterpreterSettingManager();
}
diff --git a/zeppelin-interpreter-integration/src/test/java/org/apache/zeppelin/integration/MiniZeppelin.java b/zeppelin-interpreter-integration/src/test/java/org/apache/zeppelin/integration/MiniZeppelin.java
index aebfa03..8e04d2b 100644
--- a/zeppelin-interpreter-integration/src/test/java/org/apache/zeppelin/integration/MiniZeppelin.java
+++ b/zeppelin-interpreter-integration/src/test/java/org/apache/zeppelin/integration/MiniZeppelin.java
@@ -43,16 +43,17 @@ public class MiniZeppelin {
private File notebookDir;
protected ZeppelinConfiguration conf;
- public void start() throws IOException {
+ public void start(Class clazz) throws IOException {
zeppelinHome = new File("..");
System.setProperty(ZeppelinConfiguration.ConfVars.ZEPPELIN_HOME.getVarName(),
zeppelinHome.getAbsolutePath());
- confDir = new File(zeppelinHome, "conf_" + getClass().getSimpleName());
- notebookDir = new File(zeppelinHome, "notebook_" + getClass().getSimpleName());
+ confDir = new File(zeppelinHome, "conf_" + clazz.getSimpleName());
+ notebookDir = new File(zeppelinHome, "notebook_" + clazz.getSimpleName());
confDir.mkdirs();
notebookDir.mkdirs();
LOGGER.info("ZEPPELIN_HOME: " + zeppelinHome.getAbsolutePath());
FileUtils.copyFile(new File(zeppelinHome, "conf/log4j.properties"), new File(confDir, "log4j.properties"));
+ FileUtils.copyFile(new File(zeppelinHome, "conf/log4j2.properties"), new File(confDir, "log4j2.properties"));
FileUtils.copyFile(new File(zeppelinHome, "conf/log4j_yarn_cluster.properties"), new File(confDir, "log4j_yarn_cluster.properties"));
System.setProperty(ZeppelinConfiguration.ConfVars.ZEPPELIN_CONF_DIR.getVarName(), confDir.getAbsolutePath());
System.setProperty(ZeppelinConfiguration.ConfVars.ZEPPELIN_NOTEBOOK_DIR.getVarName(), notebookDir.getAbsolutePath());
diff --git a/zeppelin-interpreter-integration/src/test/java/org/apache/zeppelin/integration/SparkIntegrationTest.java b/zeppelin-interpreter-integration/src/test/java/org/apache/zeppelin/integration/SparkIntegrationTest.java
index bb34661..b882307 100644
--- a/zeppelin-interpreter-integration/src/test/java/org/apache/zeppelin/integration/SparkIntegrationTest.java
+++ b/zeppelin-interpreter-integration/src/test/java/org/apache/zeppelin/integration/SparkIntegrationTest.java
@@ -73,7 +73,7 @@ public abstract class SparkIntegrationTest {
hadoopCluster.start();
zeppelin = new MiniZeppelin();
- zeppelin.start();
+ zeppelin.start(SparkIntegrationTest.class);
interpreterFactory = zeppelin.getInterpreterFactory();
interpreterSettingManager = zeppelin.getInterpreterSettingManager();
}
diff --git a/zeppelin-interpreter-integration/src/test/java/org/apache/zeppelin/integration/YarnInterpreterLauncherIntegrationTest.java b/zeppelin-interpreter-integration/src/test/java/org/apache/zeppelin/integration/YarnInterpreterLauncherIntegrationTest.java
index 4f1543c..5623a25 100644
--- a/zeppelin-interpreter-integration/src/test/java/org/apache/zeppelin/integration/YarnInterpreterLauncherIntegrationTest.java
+++ b/zeppelin-interpreter-integration/src/test/java/org/apache/zeppelin/integration/YarnInterpreterLauncherIntegrationTest.java
@@ -64,7 +64,7 @@ public class YarnInterpreterLauncherIntegrationTest {
hadoopCluster.start();
zeppelin = new MiniZeppelin();
- zeppelin.start();
+ zeppelin.start(YarnInterpreterLauncherIntegrationTest.class);
interpreterFactory = zeppelin.getInterpreterFactory();
interpreterSettingManager = zeppelin.getInterpreterSettingManager();
}
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 071d37c..e160206 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
@@ -20,6 +20,7 @@ package org.apache.zeppelin.conf;
import com.google.common.annotations.VisibleForTesting;
import java.io.File;
import java.io.IOException;
+import java.io.Writer;
import java.net.URI;
import java.net.URISyntaxException;
import java.net.URL;
@@ -33,6 +34,7 @@ import org.apache.commons.configuration.XMLConfiguration;
import org.apache.commons.configuration.tree.ConfigurationNode;
import org.apache.commons.exec.environment.EnvironmentUtils;
import org.apache.commons.lang3.StringUtils;
+import org.apache.zeppelin.interpreter.lifecycle.NullLifecycleManager;
import org.apache.zeppelin.util.Util;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -186,6 +188,10 @@ public class ZeppelinConfiguration extends XMLConfiguration {
conf = null;
}
+ public void setProperty(String name, String value) {
+ this.properties.put(name, value);
+ }
+
private String getStringValue(String name, String d) {
String value = this.properties.get(name);
if (value != null) {
@@ -886,6 +892,24 @@ public class ZeppelinConfiguration extends XMLConfiguration {
return properties;
}
+ @Override
+ public void save(Writer writer) throws ConfigurationException {
+ try {
+ writer.write("<configuration>\n");
+ for (Map.Entry<String, String> entry : properties.entrySet()) {
+ writer.write("<property>\n");
+ writer.write("<name>" + entry.getKey() + "</name>\n");
+ writer.write("<value>" + entry.getValue() + "</value>\n");
+ writer.write("</property>\n");
+ }
+ writer.write("</configuration>");
+ writer.close();
+ } catch (IOException e) {
+ throw new ConfigurationException(e);
+ }
+
+ }
+
/**
* Wrapper class.
*/
@@ -1018,9 +1042,9 @@ public class ZeppelinConfiguration extends XMLConfiguration {
ZEPPELIN_INTERPRETER_RPC_PORTRANGE("zeppelin.interpreter.rpc.portRange", ":"),
ZEPPELIN_INTERPRETER_LIFECYCLE_MANAGER_CLASS("zeppelin.interpreter.lifecyclemanager.class",
- "org.apache.zeppelin.interpreter.lifecycle.NullLifecycleManager"),
+ NullLifecycleManager.class.getName()),
ZEPPELIN_INTERPRETER_LIFECYCLE_MANAGER_TIMEOUT_CHECK_INTERVAL(
- "zeppelin.interpreter.lifecyclemanager.timeout.checkinterval", 6000L),
+ "zeppelin.interpreter.lifecyclemanager.timeout.checkinterval", 60000L),
ZEPPELIN_INTERPRETER_LIFECYCLE_MANAGER_TIMEOUT_THRESHOLD(
"zeppelin.interpreter.lifecyclemanager.timeout.threshold", 3600000L),
diff --git a/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/LifecycleManager.java b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/LifecycleManager.java
similarity index 55%
rename from zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/LifecycleManager.java
rename to zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/LifecycleManager.java
index f36cb0d..8f44b2f 100644
--- a/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/LifecycleManager.java
+++ b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/LifecycleManager.java
@@ -19,14 +19,24 @@
package org.apache.zeppelin.interpreter;
+import org.apache.zeppelin.conf.ZeppelinConfiguration;
+import org.apache.zeppelin.interpreter.remote.RemoteInterpreterServer;
+
/**
- * Interface for managing the lifecycle of interpreters
+ * Abstract base class for managing the lifecycle of interpreters
*/
-public interface LifecycleManager {
+public abstract class LifecycleManager {
+
+ protected ZeppelinConfiguration zConf;
+ protected RemoteInterpreterServer remoteInterpreterServer;
+
+ public LifecycleManager(ZeppelinConfiguration zConf, RemoteInterpreterServer remoteInterpreterServer) {
+ this.zConf = zConf;
+ this.remoteInterpreterServer = remoteInterpreterServer;
+ }
- void onInterpreterProcessStarted(ManagedInterpreterGroup interpreterGroup);
+ public abstract void onInterpreterProcessStarted(String interpreterGroupId);
- void onInterpreterUse(ManagedInterpreterGroup interpreterGroup,
- String sessionId);
+ public abstract void onInterpreterUse(String interpreterGroupId);
}
diff --git a/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/lifecycle/NullLifecycleManager.java b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/lifecycle/NullLifecycleManager.java
similarity index 71%
rename from zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/lifecycle/NullLifecycleManager.java
rename to zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/lifecycle/NullLifecycleManager.java
index 5a62d22..da09c0f 100644
--- a/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/lifecycle/NullLifecycleManager.java
+++ b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/lifecycle/NullLifecycleManager.java
@@ -20,24 +20,25 @@ package org.apache.zeppelin.interpreter.lifecycle;
import org.apache.zeppelin.conf.ZeppelinConfiguration;
import org.apache.zeppelin.interpreter.LifecycleManager;
-import org.apache.zeppelin.interpreter.ManagedInterpreterGroup;
+import org.apache.zeppelin.interpreter.remote.RemoteInterpreterServer;
/**
* Do nothing for the lifecycle of interpreter. User need to explicitly start/stop interpreter.
*/
-public class NullLifecycleManager implements LifecycleManager {
-
- public NullLifecycleManager(ZeppelinConfiguration zConf) {
+public class NullLifecycleManager extends LifecycleManager {
+ public NullLifecycleManager(ZeppelinConfiguration zConf,
+ RemoteInterpreterServer remoteInterpreterServer) {
+ super(zConf, remoteInterpreterServer);
}
@Override
- public void onInterpreterProcessStarted(ManagedInterpreterGroup interpreterGroup) {
+ public void onInterpreterProcessStarted(String interpreterGroupId) {
}
@Override
- public void onInterpreterUse(ManagedInterpreterGroup interpreterGroup, String sessionId) {
+ public void onInterpreterUse(String interpreterGroupId) {
}
}
diff --git a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/lifecycle/TimeoutLifecycleManager.java b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/lifecycle/TimeoutLifecycleManager.java
new file mode 100644
index 0000000..9cbb42f
--- /dev/null
+++ b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/lifecycle/TimeoutLifecycleManager.java
@@ -0,0 +1,82 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.zeppelin.interpreter.lifecycle;
+
+import org.apache.thrift.TException;
+import org.apache.zeppelin.conf.ZeppelinConfiguration;
+import org.apache.zeppelin.interpreter.LifecycleManager;
+import org.apache.zeppelin.interpreter.remote.RemoteInterpreterServer;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+
+import static java.util.concurrent.TimeUnit.MILLISECONDS;
+
+
+/**
+ * This lifecycle manager would close interpreter after it is timeout. By default, it is timeout
+ * after no using in 1 hour.
+ *
+ * For now, this class only manage the lifecycle of interpreter group (will close interpreter
+ * process after timeout). Managing the lifecycle of interpreter session could be done in future
+ * if necessary.
+ */
+public class TimeoutLifecycleManager extends LifecycleManager {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(TimeoutLifecycleManager.class);
+
+ private long lastBusyTimeInMillis;
+
+ public TimeoutLifecycleManager(ZeppelinConfiguration zConf,
+ RemoteInterpreterServer remoteInterpreterServer) {
+ super(zConf, remoteInterpreterServer);
+ long checkInterval = zConf.getLong(ZeppelinConfiguration.ConfVars
+ .ZEPPELIN_INTERPRETER_LIFECYCLE_MANAGER_TIMEOUT_CHECK_INTERVAL);
+ long timeoutThreshold = zConf.getLong(
+ ZeppelinConfiguration.ConfVars.ZEPPELIN_INTERPRETER_LIFECYCLE_MANAGER_TIMEOUT_THRESHOLD);
+ ScheduledExecutorService checkScheduler = Executors.newScheduledThreadPool(1);
+ checkScheduler.scheduleAtFixedRate(() -> {
+ if ((System.currentTimeMillis() - lastBusyTimeInMillis) > timeoutThreshold) {
+ LOGGER.info("Interpreter process idle time exceed threshold, try to stop it");
+ try {
+ remoteInterpreterServer.shutdown();
+ } catch (TException e) {
+ LOGGER.error("Fail to shutdown RemoteInterpreterServer", e);
+ }
+ } else {
+ LOGGER.debug("Check idle time of interpreter");
+ }
+ }, checkInterval, checkInterval, MILLISECONDS);
+ LOGGER.info("TimeoutLifecycleManager is started with checkInterval: " + checkInterval
+ + ", timeoutThreshold: " + timeoutThreshold);
+ }
+
+ @Override
+ public void onInterpreterProcessStarted(String interpreterGroupId) {
+ LOGGER.info("Interpreter process: {} is started", interpreterGroupId);
+ lastBusyTimeInMillis = System.currentTimeMillis();
+ }
+
+ @Override
+ public void onInterpreterUse(String interpreterGroupId) {
+ LOGGER.debug("Interpreter process: {} is used", interpreterGroupId);
+ lastBusyTimeInMillis = System.currentTimeMillis();
+ }
+}
diff --git a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreterEventClient.java b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreterEventClient.java
index 2fb0733..3eed43d 100644
--- a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreterEventClient.java
+++ b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreterEventClient.java
@@ -92,6 +92,13 @@ public class RemoteInterpreterEventClient implements ResourcePoolConnector,
});
}
+ public void unRegisterInterpreterProcess() {
+ callRemoteFunction(client -> {
+ client.unRegisterInterpreterProcess(intpGroupId);
+ return null;
+ });
+ }
+
public void sendWebUrlInfo(String webUrl) {
callRemoteFunction(client -> {
client.sendWebUrl(new WebUrlInfo(intpGroupId, webUrl));
diff --git a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreterServer.java b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreterServer.java
index 90e9aba..40e0fb9 100644
--- a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreterServer.java
+++ b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreterServer.java
@@ -22,12 +22,8 @@ import com.google.gson.reflect.TypeToken;
import org.apache.commons.lang3.exception.ExceptionUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.thrift.TException;
-import org.apache.thrift.protocol.TBinaryProtocol;
-import org.apache.thrift.protocol.TProtocol;
import org.apache.thrift.server.TThreadPoolServer;
import org.apache.thrift.transport.TServerSocket;
-import org.apache.thrift.transport.TSocket;
-import org.apache.thrift.transport.TTransport;
import org.apache.thrift.transport.TTransportException;
import org.apache.zeppelin.cluster.ClusterManagerClient;
import org.apache.zeppelin.cluster.meta.ClusterMeta;
@@ -57,11 +53,11 @@ import org.apache.zeppelin.interpreter.InterpreterResult.Code;
import org.apache.zeppelin.interpreter.InterpreterResultMessage;
import org.apache.zeppelin.interpreter.InterpreterResultMessageOutput;
import org.apache.zeppelin.interpreter.LazyOpenInterpreter;
+import org.apache.zeppelin.interpreter.LifecycleManager;
import org.apache.zeppelin.interpreter.thrift.InterpreterCompletion;
import org.apache.zeppelin.interpreter.thrift.RegisterInfo;
import org.apache.zeppelin.interpreter.thrift.RemoteApplicationResult;
import org.apache.zeppelin.interpreter.thrift.RemoteInterpreterContext;
-import org.apache.zeppelin.interpreter.thrift.RemoteInterpreterEventService;
import org.apache.zeppelin.interpreter.thrift.RemoteInterpreterResult;
import org.apache.zeppelin.interpreter.thrift.RemoteInterpreterResultMessage;
import org.apache.zeppelin.interpreter.thrift.RemoteInterpreterService;
@@ -129,6 +125,7 @@ public class RemoteInterpreterServer extends Thread
private TThreadPoolServer server;
RemoteInterpreterEventClient intpEventClient;
private DependencyResolver depLoader;
+ private LifecycleManager lifecycleManager;
private final Map<String, RunningApplication> runningApplications =
Collections.synchronizedMap(new HashMap<String, RunningApplication>());
@@ -150,14 +147,13 @@ public class RemoteInterpreterServer extends Thread
private boolean isTest;
// cluster manager client
- private ZeppelinConfiguration zconf = ZeppelinConfiguration.create();
+ private ZeppelinConfiguration zConf = ZeppelinConfiguration.create();
private ClusterManagerClient clusterManagerClient;
public RemoteInterpreterServer(String intpEventServerHost,
int intpEventServerPort,
String interpreterGroupId,
- String portRange)
- throws IOException, TTransportException {
+ String portRange) throws Exception {
this(intpEventServerHost, intpEventServerPort, portRange, interpreterGroupId, false);
}
@@ -165,8 +161,7 @@ public class RemoteInterpreterServer extends Thread
int intpEventServerPort,
String portRange,
String interpreterGroupId,
- boolean isTest)
- throws TTransportException, IOException {
+ boolean isTest) throws Exception {
LOGGER.info("Starting remote interpreter server on port {}, intpEventServerAddress: {}:{}", port,
intpEventServerHost, intpEventServerPort);
if (null != intpEventServerHost) {
@@ -197,10 +192,12 @@ public class RemoteInterpreterServer extends Thread
new TThreadPoolServer.Args(serverTransport).processor(processor));
remoteWorksResponsePool = Collections.synchronizedMap(new HashMap<String, Object>());
- if (zconf.isClusterMode()) {
- clusterManagerClient = ClusterManagerClient.getInstance(zconf);
+ if (zConf.isClusterMode()) {
+ clusterManagerClient = ClusterManagerClient.getInstance(zConf);
clusterManagerClient.start(interpreterGroupId);
}
+
+ lifecycleManager = createLifecycleManager();
}
@Override
@@ -219,7 +216,7 @@ public class RemoteInterpreterServer extends Thread
}
}
- if (zconf.isClusterMode()) {
+ if (zConf.isClusterMode()) {
// Cluster mode, discovering interpreter processes through metadata registration
// TODO (Xun): Unified use of cluster metadata for process discovery of all operating modes
// 1. Can optimize the startup logic of the process
@@ -232,6 +229,7 @@ public class RemoteInterpreterServer extends Thread
LOGGER.info("Registering interpreter process");
intpEventClient.registerInterpreterProcess(registerInfo);
LOGGER.info("Registered interpreter process");
+ lifecycleManager.onInterpreterProcessStarted(interpreterGroupId);
} catch (Exception e) {
LOGGER.error("Error while registering interpreter: {}", registerInfo, e);
try {
@@ -270,6 +268,20 @@ public class RemoteInterpreterServer extends Thread
@Override
public void shutdown() throws TException {
+
+ // unRegisterInterpreterProcess should be a sync operation (outside of shutdown thread),
+ // otherwise it would cause data mismatch between zeppelin server & interpreter process.
+ // e.g. zeppelin server start a new interpreter process, while previous interpreter process
+ // uniregister it with the same interpreter group id: flink-shared-process.
+ if (intpEventClient != null) {
+ try {
+ LOGGER.info("Unregister interpreter process");
+ intpEventClient.unRegisterInterpreterProcess();
+ } catch (Exception e) {
+ LOGGER.error("Fail to unregister remote interpreter process", e);
+ }
+ }
+
Thread shutDownThread = new Thread(() -> {
LOGGER.info("Shutting down...");
// delete interpreter cluster meta
@@ -339,8 +351,15 @@ public class RemoteInterpreterServer extends Thread
}
}
- public static void main(String[] args)
- throws TTransportException, InterruptedException, IOException {
+ private LifecycleManager createLifecycleManager() throws Exception {
+ String lifecycleManagerClass = zConf.getLifecycleManagerClass();
+ Class clazz = Class.forName(lifecycleManagerClass);
+ LOGGER.info("Creating interpreter lifecycle manager: " + lifecycleManagerClass);
+ return (LifecycleManager) clazz.getConstructor(ZeppelinConfiguration.class, RemoteInterpreterServer.class)
+ .newInstance(zConf, this);
+ }
+
+ public static void main(String[] args) throws Exception {
String zeppelinServerHost = null;
int port = Constants.ZEPPELIN_INTERPRETER_DEFAUlT_PORT;
String portRange = ":";
@@ -377,7 +396,7 @@ public class RemoteInterpreterServer extends Thread
// Submit interpreter process metadata information to cluster metadata
private void putClusterMeta() {
- if (!zconf.isClusterMode()){
+ if (!zConf.isClusterMode()){
return;
}
String nodeName = clusterManagerClient.getClusterNodeName();
@@ -396,7 +415,7 @@ public class RemoteInterpreterServer extends Thread
}
private void deleteClusterMeta() {
- if (!zconf.isClusterMode()){
+ if (!zConf.isClusterMode()){
return;
}
@@ -592,6 +611,8 @@ public class RemoteInterpreterServer extends Thread
if (LOGGER.isDebugEnabled()) {
LOGGER.debug("st:\n{}", st);
}
+ lifecycleManager.onInterpreterUse(interpreterGroupId);
+
Interpreter intp = getInterpreter(sessionId, className);
InterpreterContext context = convert(interpreterContext);
context.setInterpreterClassName(intp.getClassName());
@@ -858,6 +879,8 @@ public class RemoteInterpreterServer extends Thread
public int getProgress(String sessionId, String className,
RemoteInterpreterContext interpreterContext)
throws TException {
+ lifecycleManager.onInterpreterUse(interpreterGroupId);
+
Integer manuallyProvidedProgress = progressMap.get(interpreterContext.getParagraphId());
if (manuallyProvidedProgress != null) {
return manuallyProvidedProgress;
@@ -984,6 +1007,8 @@ public class RemoteInterpreterServer extends Thread
@Override
public String getStatus(String sessionId, String jobId)
throws TException {
+
+ lifecycleManager.onInterpreterUse(interpreterGroupId);
if (interpreterGroup == null) {
return Status.UNKNOWN.name();
}
diff --git a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/AngularObjectId.java b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/AngularObjectId.java
index 960204d..54fba0e 100644
--- a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/AngularObjectId.java
+++ b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/AngularObjectId.java
@@ -24,7 +24,7 @@
package org.apache.zeppelin.interpreter.thrift;
@SuppressWarnings({"cast", "rawtypes", "serial", "unchecked", "unused"})
-@javax.annotation.Generated(value = "Autogenerated by Thrift Compiler (0.13.0)", date = "2020-07-28")
+@javax.annotation.Generated(value = "Autogenerated by Thrift Compiler (0.13.0)", date = "2020-09-14")
public class AngularObjectId implements org.apache.thrift.TBase<AngularObjectId, AngularObjectId._Fields>, java.io.Serializable, Cloneable, Comparable<AngularObjectId> {
private static final org.apache.thrift.protocol.TStruct STRUCT_DESC = new org.apache.thrift.protocol.TStruct("AngularObjectId");
diff --git a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/AppOutputAppendEvent.java b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/AppOutputAppendEvent.java
index e245312..1652694 100644
--- a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/AppOutputAppendEvent.java
+++ b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/AppOutputAppendEvent.java
@@ -24,7 +24,7 @@
package org.apache.zeppelin.interpreter.thrift;
@SuppressWarnings({"cast", "rawtypes", "serial", "unchecked", "unused"})
-@javax.annotation.Generated(value = "Autogenerated by Thrift Compiler (0.13.0)", date = "2020-07-28")
+@javax.annotation.Generated(value = "Autogenerated by Thrift Compiler (0.13.0)", date = "2020-09-14")
public class AppOutputAppendEvent implements org.apache.thrift.TBase<AppOutputAppendEvent, AppOutputAppendEvent._Fields>, java.io.Serializable, Cloneable, Comparable<AppOutputAppendEvent> {
private static final org.apache.thrift.protocol.TStruct STRUCT_DESC = new org.apache.thrift.protocol.TStruct("AppOutputAppendEvent");
diff --git a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/AppOutputUpdateEvent.java b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/AppOutputUpdateEvent.java
index fb6cf9e..a088945 100644
--- a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/AppOutputUpdateEvent.java
+++ b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/AppOutputUpdateEvent.java
@@ -24,7 +24,7 @@
package org.apache.zeppelin.interpreter.thrift;
@SuppressWarnings({"cast", "rawtypes", "serial", "unchecked", "unused"})
-@javax.annotation.Generated(value = "Autogenerated by Thrift Compiler (0.13.0)", date = "2020-07-28")
+@javax.annotation.Generated(value = "Autogenerated by Thrift Compiler (0.13.0)", date = "2020-09-14")
public class AppOutputUpdateEvent implements org.apache.thrift.TBase<AppOutputUpdateEvent, AppOutputUpdateEvent._Fields>, java.io.Serializable, Cloneable, Comparable<AppOutputUpdateEvent> {
private static final org.apache.thrift.protocol.TStruct STRUCT_DESC = new org.apache.thrift.protocol.TStruct("AppOutputUpdateEvent");
diff --git a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/AppStatusUpdateEvent.java b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/AppStatusUpdateEvent.java
index a25cc3b..ea3f49e 100644
--- a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/AppStatusUpdateEvent.java
+++ b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/AppStatusUpdateEvent.java
@@ -24,7 +24,7 @@
package org.apache.zeppelin.interpreter.thrift;
@SuppressWarnings({"cast", "rawtypes", "serial", "unchecked", "unused"})
-@javax.annotation.Generated(value = "Autogenerated by Thrift Compiler (0.13.0)", date = "2020-07-28")
+@javax.annotation.Generated(value = "Autogenerated by Thrift Compiler (0.13.0)", date = "2020-09-14")
public class AppStatusUpdateEvent implements org.apache.thrift.TBase<AppStatusUpdateEvent, AppStatusUpdateEvent._Fields>, java.io.Serializable, Cloneable, Comparable<AppStatusUpdateEvent> {
private static final org.apache.thrift.protocol.TStruct STRUCT_DESC = new org.apache.thrift.protocol.TStruct("AppStatusUpdateEvent");
diff --git a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/InterpreterCompletion.java b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/InterpreterCompletion.java
index 8ad9da1..7a47f3c 100644
--- a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/InterpreterCompletion.java
+++ b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/InterpreterCompletion.java
@@ -24,7 +24,7 @@
package org.apache.zeppelin.interpreter.thrift;
@SuppressWarnings({"cast", "rawtypes", "serial", "unchecked", "unused"})
-@javax.annotation.Generated(value = "Autogenerated by Thrift Compiler (0.13.0)", date = "2020-07-28")
+@javax.annotation.Generated(value = "Autogenerated by Thrift Compiler (0.13.0)", date = "2020-09-14")
public class InterpreterCompletion implements org.apache.thrift.TBase<InterpreterCompletion, InterpreterCompletion._Fields>, java.io.Serializable, Cloneable, Comparable<InterpreterCompletion> {
private static final org.apache.thrift.protocol.TStruct STRUCT_DESC = new org.apache.thrift.protocol.TStruct("InterpreterCompletion");
diff --git a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/OutputAppendEvent.java b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/OutputAppendEvent.java
index a7db8e8..5e3d2de 100644
--- a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/OutputAppendEvent.java
+++ b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/OutputAppendEvent.java
@@ -24,7 +24,7 @@
package org.apache.zeppelin.interpreter.thrift;
@SuppressWarnings({"cast", "rawtypes", "serial", "unchecked", "unused"})
-@javax.annotation.Generated(value = "Autogenerated by Thrift Compiler (0.13.0)", date = "2020-07-28")
+@javax.annotation.Generated(value = "Autogenerated by Thrift Compiler (0.13.0)", date = "2020-09-14")
public class OutputAppendEvent implements org.apache.thrift.TBase<OutputAppendEvent, OutputAppendEvent._Fields>, java.io.Serializable, Cloneable, Comparable<OutputAppendEvent> {
private static final org.apache.thrift.protocol.TStruct STRUCT_DESC = new org.apache.thrift.protocol.TStruct("OutputAppendEvent");
diff --git a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/OutputUpdateAllEvent.java b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/OutputUpdateAllEvent.java
index 6d2238f..f8e0b23 100644
--- a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/OutputUpdateAllEvent.java
+++ b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/OutputUpdateAllEvent.java
@@ -24,7 +24,7 @@
package org.apache.zeppelin.interpreter.thrift;
@SuppressWarnings({"cast", "rawtypes", "serial", "unchecked", "unused"})
-@javax.annotation.Generated(value = "Autogenerated by Thrift Compiler (0.13.0)", date = "2020-07-28")
+@javax.annotation.Generated(value = "Autogenerated by Thrift Compiler (0.13.0)", date = "2020-09-14")
public class OutputUpdateAllEvent implements org.apache.thrift.TBase<OutputUpdateAllEvent, OutputUpdateAllEvent._Fields>, java.io.Serializable, Cloneable, Comparable<OutputUpdateAllEvent> {
private static final org.apache.thrift.protocol.TStruct STRUCT_DESC = new org.apache.thrift.protocol.TStruct("OutputUpdateAllEvent");
diff --git a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/OutputUpdateEvent.java b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/OutputUpdateEvent.java
index f53470c..2627b9d 100644
--- a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/OutputUpdateEvent.java
+++ b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/OutputUpdateEvent.java
@@ -24,7 +24,7 @@
package org.apache.zeppelin.interpreter.thrift;
@SuppressWarnings({"cast", "rawtypes", "serial", "unchecked", "unused"})
-@javax.annotation.Generated(value = "Autogenerated by Thrift Compiler (0.13.0)", date = "2020-07-28")
+@javax.annotation.Generated(value = "Autogenerated by Thrift Compiler (0.13.0)", date = "2020-09-14")
public class OutputUpdateEvent implements org.apache.thrift.TBase<OutputUpdateEvent, OutputUpdateEvent._Fields>, java.io.Serializable, Cloneable, Comparable<OutputUpdateEvent> {
private static final org.apache.thrift.protocol.TStruct STRUCT_DESC = new org.apache.thrift.protocol.TStruct("OutputUpdateEvent");
diff --git a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/ParagraphInfo.java b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/ParagraphInfo.java
index 804951c..deef016 100644
--- a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/ParagraphInfo.java
+++ b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/ParagraphInfo.java
@@ -24,7 +24,7 @@
package org.apache.zeppelin.interpreter.thrift;
@SuppressWarnings({"cast", "rawtypes", "serial", "unchecked", "unused"})
-@javax.annotation.Generated(value = "Autogenerated by Thrift Compiler (0.13.0)", date = "2020-07-28")
+@javax.annotation.Generated(value = "Autogenerated by Thrift Compiler (0.13.0)", date = "2020-09-14")
public class ParagraphInfo implements org.apache.thrift.TBase<ParagraphInfo, ParagraphInfo._Fields>, java.io.Serializable, Cloneable, Comparable<ParagraphInfo> {
private static final org.apache.thrift.protocol.TStruct STRUCT_DESC = new org.apache.thrift.protocol.TStruct("ParagraphInfo");
diff --git a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/RegisterInfo.java b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/RegisterInfo.java
index d2d19c9..4a3fb80 100644
--- a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/RegisterInfo.java
+++ b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/RegisterInfo.java
@@ -24,7 +24,7 @@
package org.apache.zeppelin.interpreter.thrift;
@SuppressWarnings({"cast", "rawtypes", "serial", "unchecked", "unused"})
-@javax.annotation.Generated(value = "Autogenerated by Thrift Compiler (0.13.0)", date = "2020-07-28")
+@javax.annotation.Generated(value = "Autogenerated by Thrift Compiler (0.13.0)", date = "2020-09-14")
public class RegisterInfo implements org.apache.thrift.TBase<RegisterInfo, RegisterInfo._Fields>, java.io.Serializable, Cloneable, Comparable<RegisterInfo> {
private static final org.apache.thrift.protocol.TStruct STRUCT_DESC = new org.apache.thrift.protocol.TStruct("RegisterInfo");
diff --git a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/RemoteApplicationResult.java b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/RemoteApplicationResult.java
index 5e5434a..3120cb5 100644
--- a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/RemoteApplicationResult.java
+++ b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/RemoteApplicationResult.java
@@ -24,7 +24,7 @@
package org.apache.zeppelin.interpreter.thrift;
@SuppressWarnings({"cast", "rawtypes", "serial", "unchecked", "unused"})
-@javax.annotation.Generated(value = "Autogenerated by Thrift Compiler (0.13.0)", date = "2020-07-28")
+@javax.annotation.Generated(value = "Autogenerated by Thrift Compiler (0.13.0)", date = "2020-09-14")
public class RemoteApplicationResult implements org.apache.thrift.TBase<RemoteApplicationResult, RemoteApplicationResult._Fields>, java.io.Serializable, Cloneable, Comparable<RemoteApplicationResult> {
private static final org.apache.thrift.protocol.TStruct STRUCT_DESC = new org.apache.thrift.protocol.TStruct("RemoteApplicationResult");
diff --git a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/RemoteInterpreterContext.java b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/RemoteInterpreterContext.java
index a39e099..2667534 100644
--- a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/RemoteInterpreterContext.java
+++ b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/RemoteInterpreterContext.java
@@ -24,7 +24,7 @@
package org.apache.zeppelin.interpreter.thrift;
@SuppressWarnings({"cast", "rawtypes", "serial", "unchecked", "unused"})
-@javax.annotation.Generated(value = "Autogenerated by Thrift Compiler (0.13.0)", date = "2020-07-28")
+@javax.annotation.Generated(value = "Autogenerated by Thrift Compiler (0.13.0)", date = "2020-09-14")
public class RemoteInterpreterContext implements org.apache.thrift.TBase<RemoteInterpreterContext, RemoteInterpreterContext._Fields>, java.io.Serializable, Cloneable, Comparable<RemoteInterpreterContext> {
private static final org.apache.thrift.protocol.TStruct STRUCT_DESC = new org.apache.thrift.protocol.TStruct("RemoteInterpreterContext");
diff --git a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/RemoteInterpreterEvent.java b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/RemoteInterpreterEvent.java
index c7153c3..2d0602a 100644
--- a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/RemoteInterpreterEvent.java
+++ b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/RemoteInterpreterEvent.java
@@ -24,7 +24,7 @@
package org.apache.zeppelin.interpreter.thrift;
@SuppressWarnings({"cast", "rawtypes", "serial", "unchecked", "unused"})
-@javax.annotation.Generated(value = "Autogenerated by Thrift Compiler (0.13.0)", date = "2020-07-28")
+@javax.annotation.Generated(value = "Autogenerated by Thrift Compiler (0.13.0)", date = "2020-09-14")
public class RemoteInterpreterEvent implements org.apache.thrift.TBase<RemoteInterpreterEvent, RemoteInterpreterEvent._Fields>, java.io.Serializable, Cloneable, Comparable<RemoteInterpreterEvent> {
private static final org.apache.thrift.protocol.TStruct STRUCT_DESC = new org.apache.thrift.protocol.TStruct("RemoteInterpreterEvent");
diff --git a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/RemoteInterpreterEventService.java b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/RemoteInterpreterEventService.java
index e59f9f0..63c1175 100644
--- a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/RemoteInterpreterEventService.java
+++ b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/RemoteInterpreterEventService.java
@@ -24,13 +24,15 @@
package org.apache.zeppelin.interpreter.thrift;
@SuppressWarnings({"cast", "rawtypes", "serial", "unchecked", "unused"})
-@javax.annotation.Generated(value = "Autogenerated by Thrift Compiler (0.13.0)", date = "2020-07-28")
+@javax.annotation.Generated(value = "Autogenerated by Thrift Compiler (0.13.0)", date = "2020-09-14")
public class RemoteInterpreterEventService {
public interface Iface {
public void registerInterpreterProcess(RegisterInfo registerInfo) throws org.apache.thrift.TException;
+ public void unRegisterInterpreterProcess(java.lang.String intpGroupId) throws org.apache.thrift.TException;
+
public void appendOutput(OutputAppendEvent event) throws org.apache.thrift.TException;
public void updateOutput(OutputUpdateEvent event) throws org.apache.thrift.TException;
@@ -73,6 +75,8 @@ public class RemoteInterpreterEventService {
public void registerInterpreterProcess(RegisterInfo registerInfo, org.apache.thrift.async.AsyncMethodCallback<Void> resultHandler) throws org.apache.thrift.TException;
+ public void unRegisterInterpreterProcess(java.lang.String intpGroupId, org.apache.thrift.async.AsyncMethodCallback<Void> resultHandler) throws org.apache.thrift.TException;
+
public void appendOutput(OutputAppendEvent event, org.apache.thrift.async.AsyncMethodCallback<Void> resultHandler) throws org.apache.thrift.TException;
public void updateOutput(OutputUpdateEvent event, org.apache.thrift.async.AsyncMethodCallback<Void> resultHandler) throws org.apache.thrift.TException;
@@ -151,6 +155,26 @@ public class RemoteInterpreterEventService {
return;
}
+ public void unRegisterInterpreterProcess(java.lang.String intpGroupId) throws org.apache.thrift.TException
+ {
+ send_unRegisterInterpreterProcess(intpGroupId);
+ recv_unRegisterInterpreterProcess();
+ }
+
+ public void send_unRegisterInterpreterProcess(java.lang.String intpGroupId) throws org.apache.thrift.TException
+ {
+ unRegisterInterpreterProcess_args args = new unRegisterInterpreterProcess_args();
+ args.setIntpGroupId(intpGroupId);
+ sendBase("unRegisterInterpreterProcess", args);
+ }
+
+ public void recv_unRegisterInterpreterProcess() throws org.apache.thrift.TException
+ {
+ unRegisterInterpreterProcess_result result = new unRegisterInterpreterProcess_result();
+ receiveBase(result, "unRegisterInterpreterProcess");
+ return;
+ }
+
public void appendOutput(OutputAppendEvent event) throws org.apache.thrift.TException
{
send_appendOutput(event);
@@ -587,6 +611,38 @@ public class RemoteInterpreterEventService {
}
}
+ public void unRegisterInterpreterProcess(java.lang.String intpGroupId, org.apache.thrift.async.AsyncMethodCallback<Void> resultHandler) throws org.apache.thrift.TException {
+ checkReady();
+ unRegisterInterpreterProcess_call method_call = new unRegisterInterpreterProcess_call(intpGroupId, resultHandler, this, ___protocolFactory, ___transport);
+ this.___currentMethod = method_call;
+ ___manager.call(method_call);
+ }
+
+ public static class unRegisterInterpreterProcess_call extends org.apache.thrift.async.TAsyncMethodCall<Void> {
+ private java.lang.String intpGroupId;
+ public unRegisterInterpreterProcess_call(java.lang.String intpGroupId, org.apache.thrift.async.AsyncMethodCallback<Void> resultHandler, org.apache.thrift.async.TAsyncClient client, org.apache.thrift.protocol.TProtocolFactory protocolFactory, org.apache.thrift.transport.TNonblockingTransport transport) throws org.apache.thrift.TException {
+ super(client, protocolFactory, transport, resultHandler, false);
+ this.intpGroupId = intpGroupId;
+ }
+
+ public void write_args(org.apache.thrift.protocol.TProtocol prot) throws org.apache.thrift.TException {
+ prot.writeMessageBegin(new org.apache.thrift.protocol.TMessage("unRegisterInterpreterProcess", org.apache.thrift.protocol.TMessageType.CALL, 0));
+ unRegisterInterpreterProcess_args args = new unRegisterInterpreterProcess_args();
+ args.setIntpGroupId(intpGroupId);
+ args.write(prot);
+ prot.writeMessageEnd();
+ }
+
+ public Void getResult() throws org.apache.thrift.TException {
+ if (getState() != org.apache.thrift.async.TAsyncMethodCall.State.RESPONSE_READ) {
+ throw new java.lang.IllegalStateException("Method call not finished!");
+ }
+ org.apache.thrift.transport.TMemoryInputTransport memoryTransport = new org.apache.thrift.transport.TMemoryInputTransport(getFrameBuffer().array());
+ org.apache.thrift.protocol.TProtocol prot = client.getProtocolFactory().getProtocol(memoryTransport);
+ return null;
+ }
+ }
+
public void appendOutput(OutputAppendEvent event, org.apache.thrift.async.AsyncMethodCallback<Void> resultHandler) throws org.apache.thrift.TException {
checkReady();
appendOutput_call method_call = new appendOutput_call(event, resultHandler, this, ___protocolFactory, ___transport);
@@ -1210,6 +1266,7 @@ public class RemoteInterpreterEventService {
private static <I extends Iface> java.util.Map<java.lang.String, org.apache.thrift.ProcessFunction<I, ? extends org.apache.thrift.TBase>> getProcessMap(java.util.Map<java.lang.String, org.apache.thrift.ProcessFunction<I, ? extends org.apache.thrift.TBase>> processMap) {
processMap.put("registerInterpreterProcess", new registerInterpreterProcess());
+ processMap.put("unRegisterInterpreterProcess", new unRegisterInterpreterProcess());
processMap.put("appendOutput", new appendOutput());
processMap.put("updateOutput", new updateOutput());
processMap.put("updateAllOutput", new updateAllOutput());
@@ -1256,6 +1313,31 @@ public class RemoteInterpreterEventService {
}
}
+ public static class unRegisterInterpreterProcess<I extends Iface> extends org.apache.thrift.ProcessFunction<I, unRegisterInterpreterProcess_args> {
+ public unRegisterInterpreterProcess() {
+ super("unRegisterInterpreterProcess");
+ }
+
+ public unRegisterInterpreterProcess_args getEmptyArgsInstance() {
+ return new unRegisterInterpreterProcess_args();
+ }
+
+ protected boolean isOneway() {
+ return false;
+ }
+
+ @Override
+ protected boolean rethrowUnhandledExceptions() {
+ return false;
+ }
+
+ public unRegisterInterpreterProcess_result getResult(I iface, unRegisterInterpreterProcess_args args) throws org.apache.thrift.TException {
+ unRegisterInterpreterProcess_result result = new unRegisterInterpreterProcess_result();
+ iface.unRegisterInterpreterProcess(args.intpGroupId);
+ return result;
+ }
+ }
+
public static class appendOutput<I extends Iface> extends org.apache.thrift.ProcessFunction<I, appendOutput_args> {
public appendOutput() {
super("appendOutput");
@@ -1724,6 +1806,7 @@ public class RemoteInterpreterEventService {
private static <I extends AsyncIface> java.util.Map<java.lang.String, org.apache.thrift.AsyncProcessFunction<I, ? extends org.apache.thrift.TBase,?>> getProcessMap(java.util.Map<java.lang.String, org.apache.thrift.AsyncProcessFunction<I, ? extends org.apache.thrift.TBase, ?>> processMap) {
processMap.put("registerInterpreterProcess", new registerInterpreterProcess());
+ processMap.put("unRegisterInterpreterProcess", new unRegisterInterpreterProcess());
processMap.put("appendOutput", new appendOutput());
processMap.put("updateOutput", new updateOutput());
processMap.put("updateAllOutput", new updateAllOutput());
@@ -1805,6 +1888,66 @@ public class RemoteInterpreterEventService {
}
}
+ public static class unRegisterInterpreterProcess<I extends AsyncIface> extends org.apache.thrift.AsyncProcessFunction<I, unRegisterInterpreterProcess_args, Void> {
+ public unRegisterInterpreterProcess() {
+ super("unRegisterInterpreterProcess");
+ }
+
+ public unRegisterInterpreterProcess_args getEmptyArgsInstance() {
+ return new unRegisterInterpreterProcess_args();
+ }
+
+ public org.apache.thrift.async.AsyncMethodCallback<Void> getResultHandler(final org.apache.thrift.server.AbstractNonblockingServer.AsyncFrameBuffer fb, final int seqid) {
+ final org.apache.thrift.AsyncProcessFunction fcall = this;
+ return new org.apache.thrift.async.AsyncMethodCallback<Void>() {
+ public void onComplete(Void o) {
+ unRegisterInterpreterProcess_result result = new unRegisterInterpreterProcess_result();
+ try {
+ fcall.sendResponse(fb, result, org.apache.thrift.protocol.TMessageType.REPLY,seqid);
+ } catch (org.apache.thrift.transport.TTransportException e) {
+ _LOGGER.error("TTransportException writing to internal frame buffer", e);
+ fb.close();
+ } catch (java.lang.Exception e) {
+ _LOGGER.error("Exception writing to internal frame buffer", e);
+ onError(e);
+ }
+ }
+ public void onError(java.lang.Exception e) {
+ byte msgType = org.apache.thrift.protocol.TMessageType.REPLY;
+ org.apache.thrift.TSerializable msg;
+ unRegisterInterpreterProcess_result result = new unRegisterInterpreterProcess_result();
+ if (e instanceof org.apache.thrift.transport.TTransportException) {
+ _LOGGER.error("TTransportException inside handler", e);
+ fb.close();
+ return;
+ } else if (e instanceof org.apache.thrift.TApplicationException) {
+ _LOGGER.error("TApplicationException inside handler", e);
+ msgType = org.apache.thrift.protocol.TMessageType.EXCEPTION;
+ msg = (org.apache.thrift.TApplicationException)e;
+ } else {
+ _LOGGER.error("Exception inside handler", e);
+ msgType = org.apache.thrift.protocol.TMessageType.EXCEPTION;
+ msg = new org.apache.thrift.TApplicationException(org.apache.thrift.TApplicationException.INTERNAL_ERROR, e.getMessage());
+ }
+ try {
+ fcall.sendResponse(fb,msg,msgType,seqid);
+ } catch (java.lang.Exception ex) {
+ _LOGGER.error("Exception writing to internal frame buffer", ex);
+ fb.close();
+ }
+ }
+ };
+ }
+
+ protected boolean isOneway() {
+ return false;
+ }
+
+ public void start(I iface, unRegisterInterpreterProcess_args args, org.apache.thrift.async.AsyncMethodCallback<Void> resultHandler) throws org.apache.thrift.TException {
+ iface.unRegisterInterpreterProcess(args.intpGroupId,resultHandler);
+ }
+ }
+
public static class appendOutput<I extends AsyncIface> extends org.apache.thrift.AsyncProcessFunction<I, appendOutput_args, Void> {
public appendOutput() {
super("appendOutput");
@@ -3521,6 +3664,627 @@ public class RemoteInterpreterEventService {
}
}
+ public static class unRegisterInterpreterProcess_args implements org.apache.thrift.TBase<unRegisterInterpreterProcess_args, unRegisterInterpreterProcess_args._Fields>, java.io.Serializable, Cloneable, Comparable<unRegisterInterpreterProcess_args> {
+ private static final org.apache.thrift.protocol.TStruct STRUCT_DESC = new org.apache.thrift.protocol.TStruct("unRegisterInterpreterProcess_args");
+
+ private static final org.apache.thrift.protocol.TField INTP_GROUP_ID_FIELD_DESC = new org.apache.thrift.protocol.TField("intpGroupId", org.apache.thrift.protocol.TType.STRING, (short)1);
+
+ private static final org.apache.thrift.scheme.SchemeFactory STANDARD_SCHEME_FACTORY = new unRegisterInterpreterProcess_argsStandardSchemeFactory();
+ private static final org.apache.thrift.scheme.SchemeFactory TUPLE_SCHEME_FACTORY = new unRegisterInterpreterProcess_argsTupleSchemeFactory();
+
+ public @org.apache.thrift.annotation.Nullable java.lang.String intpGroupId; // required
+
+ /** The set of fields this struct contains, along with convenience methods for finding and manipulating them. */
+ public enum _Fields implements org.apache.thrift.TFieldIdEnum {
+ INTP_GROUP_ID((short)1, "intpGroupId");
+
+ private static final java.util.Map<java.lang.String, _Fields> byName = new java.util.HashMap<java.lang.String, _Fields>();
+
+ static {
+ for (_Fields field : java.util.EnumSet.allOf(_Fields.class)) {
+ byName.put(field.getFieldName(), field);
+ }
+ }
+
+ /**
+ * Find the _Fields constant that matches fieldId, or null if its not found.
+ */
+ @org.apache.thrift.annotation.Nullable
+ public static _Fields findByThriftId(int fieldId) {
+ switch(fieldId) {
+ case 1: // INTP_GROUP_ID
+ return INTP_GROUP_ID;
+ default:
+ return null;
+ }
+ }
+
+ /**
+ * Find the _Fields constant that matches fieldId, throwing an exception
+ * if it is not found.
+ */
+ public static _Fields findByThriftIdOrThrow(int fieldId) {
+ _Fields fields = findByThriftId(fieldId);
+ if (fields == null) throw new java.lang.IllegalArgumentException("Field " + fieldId + " doesn't exist!");
+ return fields;
+ }
+
+ /**
+ * Find the _Fields constant that matches name, or null if its not found.
+ */
+ @org.apache.thrift.annotation.Nullable
+ public static _Fields findByName(java.lang.String name) {
+ return byName.get(name);
+ }
+
+ private final short _thriftId;
+ private final java.lang.String _fieldName;
+
+ _Fields(short thriftId, java.lang.String fieldName) {
+ _thriftId = thriftId;
+ _fieldName = fieldName;
+ }
+
+ public short getThriftFieldId() {
+ return _thriftId;
+ }
+
+ public java.lang.String getFieldName() {
+ return _fieldName;
+ }
+ }
+
+ // isset id assignments
+ public static final java.util.Map<_Fields, org.apache.thrift.meta_data.FieldMetaData> metaDataMap;
+ static {
+ java.util.Map<_Fields, org.apache.thrift.meta_data.FieldMetaData> tmpMap = new java.util.EnumMap<_Fields, org.apache.thrift.meta_data.FieldMetaData>(_Fields.class);
+ tmpMap.put(_Fields.INTP_GROUP_ID, new org.apache.thrift.meta_data.FieldMetaData("intpGroupId", org.apache.thrift.TFieldRequirementType.DEFAULT,
+ new org.apache.thrift.meta_data.FieldValueMetaData(org.apache.thrift.protocol.TType.STRING)));
+ metaDataMap = java.util.Collections.unmodifiableMap(tmpMap);
+ org.apache.thrift.meta_data.FieldMetaData.addStructMetaDataMap(unRegisterInterpreterProcess_args.class, metaDataMap);
+ }
+
+ public unRegisterInterpreterProcess_args() {
+ }
+
+ public unRegisterInterpreterProcess_args(
+ java.lang.String intpGroupId)
+ {
+ this();
+ this.intpGroupId = intpGroupId;
+ }
+
+ /**
+ * Performs a deep copy on <i>other</i>.
+ */
+ public unRegisterInterpreterProcess_args(unRegisterInterpreterProcess_args other) {
+ if (other.isSetIntpGroupId()) {
+ this.intpGroupId = other.intpGroupId;
+ }
+ }
+
+ public unRegisterInterpreterProcess_args deepCopy() {
+ return new unRegisterInterpreterProcess_args(this);
+ }
+
+ @Override
+ public void clear() {
+ this.intpGroupId = null;
+ }
+
+ @org.apache.thrift.annotation.Nullable
+ public java.lang.String getIntpGroupId() {
+ return this.intpGroupId;
+ }
+
+ public unRegisterInterpreterProcess_args setIntpGroupId(@org.apache.thrift.annotation.Nullable java.lang.String intpGroupId) {
+ this.intpGroupId = intpGroupId;
+ return this;
+ }
+
+ public void unsetIntpGroupId() {
+ this.intpGroupId = null;
+ }
+
+ /** Returns true if field intpGroupId is set (has been assigned a value) and false otherwise */
+ public boolean isSetIntpGroupId() {
+ return this.intpGroupId != null;
+ }
+
+ public void setIntpGroupIdIsSet(boolean value) {
+ if (!value) {
+ this.intpGroupId = null;
+ }
+ }
+
+ public void setFieldValue(_Fields field, @org.apache.thrift.annotation.Nullable java.lang.Object value) {
+ switch (field) {
+ case INTP_GROUP_ID:
+ if (value == null) {
+ unsetIntpGroupId();
+ } else {
+ setIntpGroupId((java.lang.String)value);
+ }
+ break;
+
+ }
+ }
+
+ @org.apache.thrift.annotation.Nullable
+ public java.lang.Object getFieldValue(_Fields field) {
+ switch (field) {
+ case INTP_GROUP_ID:
+ return getIntpGroupId();
+
+ }
+ throw new java.lang.IllegalStateException();
+ }
+
+ /** Returns true if field corresponding to fieldID is set (has been assigned a value) and false otherwise */
+ public boolean isSet(_Fields field) {
+ if (field == null) {
+ throw new java.lang.IllegalArgumentException();
+ }
+
+ switch (field) {
+ case INTP_GROUP_ID:
+ return isSetIntpGroupId();
+ }
+ throw new java.lang.IllegalStateException();
+ }
+
+ @Override
+ public boolean equals(java.lang.Object that) {
+ if (that == null)
+ return false;
+ if (that instanceof unRegisterInterpreterProcess_args)
+ return this.equals((unRegisterInterpreterProcess_args)that);
+ return false;
+ }
+
+ public boolean equals(unRegisterInterpreterProcess_args that) {
+ if (that == null)
+ return false;
+ if (this == that)
+ return true;
+
+ boolean this_present_intpGroupId = true && this.isSetIntpGroupId();
+ boolean that_present_intpGroupId = true && that.isSetIntpGroupId();
+ if (this_present_intpGroupId || that_present_intpGroupId) {
+ if (!(this_present_intpGroupId && that_present_intpGroupId))
+ return false;
+ if (!this.intpGroupId.equals(that.intpGroupId))
+ return false;
+ }
+
+ return true;
+ }
+
+ @Override
+ public int hashCode() {
+ int hashCode = 1;
+
+ hashCode = hashCode * 8191 + ((isSetIntpGroupId()) ? 131071 : 524287);
+ if (isSetIntpGroupId())
+ hashCode = hashCode * 8191 + intpGroupId.hashCode();
+
+ return hashCode;
+ }
+
+ @Override
+ public int compareTo(unRegisterInterpreterProcess_args other) {
+ if (!getClass().equals(other.getClass())) {
+ return getClass().getName().compareTo(other.getClass().getName());
+ }
+
+ int lastComparison = 0;
+
+ lastComparison = java.lang.Boolean.valueOf(isSetIntpGroupId()).compareTo(other.isSetIntpGroupId());
+ if (lastComparison != 0) {
+ return lastComparison;
+ }
+ if (isSetIntpGroupId()) {
+ lastComparison = org.apache.thrift.TBaseHelper.compareTo(this.intpGroupId, other.intpGroupId);
+ if (lastComparison != 0) {
+ return lastComparison;
+ }
+ }
+ return 0;
+ }
+
+ @org.apache.thrift.annotation.Nullable
+ public _Fields fieldForId(int fieldId) {
+ return _Fields.findByThriftId(fieldId);
+ }
+
+ public void read(org.apache.thrift.protocol.TProtocol iprot) throws org.apache.thrift.TException {
+ scheme(iprot).read(iprot, this);
+ }
+
+ public void write(org.apache.thrift.protocol.TProtocol oprot) throws org.apache.thrift.TException {
+ scheme(oprot).write(oprot, this);
+ }
+
+ @Override
+ public java.lang.String toString() {
+ java.lang.StringBuilder sb = new java.lang.StringBuilder("unRegisterInterpreterProcess_args(");
+ boolean first = true;
+
+ sb.append("intpGroupId:");
+ if (this.intpGroupId == null) {
+ sb.append("null");
+ } else {
+ sb.append(this.intpGroupId);
+ }
+ first = false;
+ sb.append(")");
+ return sb.toString();
+ }
+
+ public void validate() throws org.apache.thrift.TException {
+ // check for required fields
+ // check for sub-struct validity
+ }
+
+ private void writeObject(java.io.ObjectOutputStream out) throws java.io.IOException {
+ try {
+ write(new org.apache.thrift.protocol.TCompactProtocol(new org.apache.thrift.transport.TIOStreamTransport(out)));
+ } catch (org.apache.thrift.TException te) {
+ throw new java.io.IOException(te);
+ }
+ }
+
+ private void readObject(java.io.ObjectInputStream in) throws java.io.IOException, java.lang.ClassNotFoundException {
+ try {
+ read(new org.apache.thrift.protocol.TCompactProtocol(new org.apache.thrift.transport.TIOStreamTransport(in)));
+ } catch (org.apache.thrift.TException te) {
+ throw new java.io.IOException(te);
+ }
+ }
+
+ private static class unRegisterInterpreterProcess_argsStandardSchemeFactory implements org.apache.thrift.scheme.SchemeFactory {
+ public unRegisterInterpreterProcess_argsStandardScheme getScheme() {
+ return new unRegisterInterpreterProcess_argsStandardScheme();
+ }
+ }
+
+ private static class unRegisterInterpreterProcess_argsStandardScheme extends org.apache.thrift.scheme.StandardScheme<unRegisterInterpreterProcess_args> {
+
+ public void read(org.apache.thrift.protocol.TProtocol iprot, unRegisterInterpreterProcess_args struct) throws org.apache.thrift.TException {
+ org.apache.thrift.protocol.TField schemeField;
+ iprot.readStructBegin();
+ while (true)
+ {
+ schemeField = iprot.readFieldBegin();
+ if (schemeField.type == org.apache.thrift.protocol.TType.STOP) {
+ break;
+ }
+ switch (schemeField.id) {
+ case 1: // INTP_GROUP_ID
+ if (schemeField.type == org.apache.thrift.protocol.TType.STRING) {
+ struct.intpGroupId = iprot.readString();
+ struct.setIntpGroupIdIsSet(true);
+ } else {
+ org.apache.thrift.protocol.TProtocolUtil.skip(iprot, schemeField.type);
+ }
+ break;
+ default:
+ org.apache.thrift.protocol.TProtocolUtil.skip(iprot, schemeField.type);
+ }
+ iprot.readFieldEnd();
+ }
+ iprot.readStructEnd();
+
+ // check for required fields of primitive type, which can't be checked in the validate method
+ struct.validate();
+ }
+
+ public void write(org.apache.thrift.protocol.TProtocol oprot, unRegisterInterpreterProcess_args struct) throws org.apache.thrift.TException {
+ struct.validate();
+
+ oprot.writeStructBegin(STRUCT_DESC);
+ if (struct.intpGroupId != null) {
+ oprot.writeFieldBegin(INTP_GROUP_ID_FIELD_DESC);
+ oprot.writeString(struct.intpGroupId);
+ oprot.writeFieldEnd();
+ }
+ oprot.writeFieldStop();
+ oprot.writeStructEnd();
+ }
+
+ }
+
+ private static class unRegisterInterpreterProcess_argsTupleSchemeFactory implements org.apache.thrift.scheme.SchemeFactory {
+ public unRegisterInterpreterProcess_argsTupleScheme getScheme() {
+ return new unRegisterInterpreterProcess_argsTupleScheme();
+ }
+ }
+
+ private static class unRegisterInterpreterProcess_argsTupleScheme extends org.apache.thrift.scheme.TupleScheme<unRegisterInterpreterProcess_args> {
+
+ @Override
+ public void write(org.apache.thrift.protocol.TProtocol prot, unRegisterInterpreterProcess_args struct) throws org.apache.thrift.TException {
+ org.apache.thrift.protocol.TTupleProtocol oprot = (org.apache.thrift.protocol.TTupleProtocol) prot;
+ java.util.BitSet optionals = new java.util.BitSet();
+ if (struct.isSetIntpGroupId()) {
+ optionals.set(0);
+ }
+ oprot.writeBitSet(optionals, 1);
+ if (struct.isSetIntpGroupId()) {
+ oprot.writeString(struct.intpGroupId);
+ }
+ }
+
+ @Override
+ public void read(org.apache.thrift.protocol.TProtocol prot, unRegisterInterpreterProcess_args struct) throws org.apache.thrift.TException {
+ org.apache.thrift.protocol.TTupleProtocol iprot = (org.apache.thrift.protocol.TTupleProtocol) prot;
+ java.util.BitSet incoming = iprot.readBitSet(1);
+ if (incoming.get(0)) {
+ struct.intpGroupId = iprot.readString();
+ struct.setIntpGroupIdIsSet(true);
+ }
+ }
+ }
+
+ private static <S extends org.apache.thrift.scheme.IScheme> S scheme(org.apache.thrift.protocol.TProtocol proto) {
+ return (org.apache.thrift.scheme.StandardScheme.class.equals(proto.getScheme()) ? STANDARD_SCHEME_FACTORY : TUPLE_SCHEME_FACTORY).getScheme();
+ }
+ }
+
+ public static class unRegisterInterpreterProcess_result implements org.apache.thrift.TBase<unRegisterInterpreterProcess_result, unRegisterInterpreterProcess_result._Fields>, java.io.Serializable, Cloneable, Comparable<unRegisterInterpreterProcess_result> {
+ private static final org.apache.thrift.protocol.TStruct STRUCT_DESC = new org.apache.thrift.protocol.TStruct("unRegisterInterpreterProcess_result");
+
+
+ private static final org.apache.thrift.scheme.SchemeFactory STANDARD_SCHEME_FACTORY = new unRegisterInterpreterProcess_resultStandardSchemeFactory();
+ private static final org.apache.thrift.scheme.SchemeFactory TUPLE_SCHEME_FACTORY = new unRegisterInterpreterProcess_resultTupleSchemeFactory();
+
+
+ /** The set of fields this struct contains, along with convenience methods for finding and manipulating them. */
+ public enum _Fields implements org.apache.thrift.TFieldIdEnum {
+;
+
+ private static final java.util.Map<java.lang.String, _Fields> byName = new java.util.HashMap<java.lang.String, _Fields>();
+
+ static {
+ for (_Fields field : java.util.EnumSet.allOf(_Fields.class)) {
+ byName.put(field.getFieldName(), field);
+ }
+ }
+
+ /**
+ * Find the _Fields constant that matches fieldId, or null if its not found.
+ */
+ @org.apache.thrift.annotation.Nullable
+ public static _Fields findByThriftId(int fieldId) {
+ switch(fieldId) {
+ default:
+ return null;
+ }
+ }
+
+ /**
+ * Find the _Fields constant that matches fieldId, throwing an exception
+ * if it is not found.
+ */
+ public static _Fields findByThriftIdOrThrow(int fieldId) {
+ _Fields fields = findByThriftId(fieldId);
+ if (fields == null) throw new java.lang.IllegalArgumentException("Field " + fieldId + " doesn't exist!");
+ return fields;
+ }
+
+ /**
+ * Find the _Fields constant that matches name, or null if its not found.
+ */
+ @org.apache.thrift.annotation.Nullable
+ public static _Fields findByName(java.lang.String name) {
+ return byName.get(name);
+ }
+
+ private final short _thriftId;
+ private final java.lang.String _fieldName;
+
+ _Fields(short thriftId, java.lang.String fieldName) {
+ _thriftId = thriftId;
+ _fieldName = fieldName;
+ }
+
+ public short getThriftFieldId() {
+ return _thriftId;
+ }
+
+ public java.lang.String getFieldName() {
+ return _fieldName;
+ }
+ }
+ public static final java.util.Map<_Fields, org.apache.thrift.meta_data.FieldMetaData> metaDataMap;
+ static {
+ java.util.Map<_Fields, org.apache.thrift.meta_data.FieldMetaData> tmpMap = new java.util.EnumMap<_Fields, org.apache.thrift.meta_data.FieldMetaData>(_Fields.class);
+ metaDataMap = java.util.Collections.unmodifiableMap(tmpMap);
+ org.apache.thrift.meta_data.FieldMetaData.addStructMetaDataMap(unRegisterInterpreterProcess_result.class, metaDataMap);
+ }
+
+ public unRegisterInterpreterProcess_result() {
+ }
+
+ /**
+ * Performs a deep copy on <i>other</i>.
+ */
+ public unRegisterInterpreterProcess_result(unRegisterInterpreterProcess_result other) {
+ }
+
+ public unRegisterInterpreterProcess_result deepCopy() {
+ return new unRegisterInterpreterProcess_result(this);
+ }
+
+ @Override
+ public void clear() {
+ }
+
+ public void setFieldValue(_Fields field, @org.apache.thrift.annotation.Nullable java.lang.Object value) {
+ switch (field) {
+ }
+ }
+
+ @org.apache.thrift.annotation.Nullable
+ public java.lang.Object getFieldValue(_Fields field) {
+ switch (field) {
+ }
+ throw new java.lang.IllegalStateException();
+ }
+
+ /** Returns true if field corresponding to fieldID is set (has been assigned a value) and false otherwise */
+ public boolean isSet(_Fields field) {
+ if (field == null) {
+ throw new java.lang.IllegalArgumentException();
+ }
+
+ switch (field) {
+ }
+ throw new java.lang.IllegalStateException();
+ }
+
+ @Override
+ public boolean equals(java.lang.Object that) {
+ if (that == null)
+ return false;
+ if (that instanceof unRegisterInterpreterProcess_result)
+ return this.equals((unRegisterInterpreterProcess_result)that);
+ return false;
+ }
+
+ public boolean equals(unRegisterInterpreterProcess_result that) {
+ if (that == null)
+ return false;
+ if (this == that)
+ return true;
+
+ return true;
+ }
+
+ @Override
+ public int hashCode() {
+ int hashCode = 1;
+
+ return hashCode;
+ }
+
+ @Override
+ public int compareTo(unRegisterInterpreterProcess_result other) {
+ if (!getClass().equals(other.getClass())) {
+ return getClass().getName().compareTo(other.getClass().getName());
+ }
+
+ int lastComparison = 0;
+
+ return 0;
+ }
+
+ @org.apache.thrift.annotation.Nullable
+ public _Fields fieldForId(int fieldId) {
+ return _Fields.findByThriftId(fieldId);
+ }
+
+ public void read(org.apache.thrift.protocol.TProtocol iprot) throws org.apache.thrift.TException {
+ scheme(iprot).read(iprot, this);
+ }
+
+ public void write(org.apache.thrift.protocol.TProtocol oprot) throws org.apache.thrift.TException {
+ scheme(oprot).write(oprot, this);
+ }
+
+ @Override
+ public java.lang.String toString() {
+ java.lang.StringBuilder sb = new java.lang.StringBuilder("unRegisterInterpreterProcess_result(");
+ boolean first = true;
+
+ sb.append(")");
+ return sb.toString();
+ }
+
+ public void validate() throws org.apache.thrift.TException {
+ // check for required fields
+ // check for sub-struct validity
+ }
+
+ private void writeObject(java.io.ObjectOutputStream out) throws java.io.IOException {
+ try {
+ write(new org.apache.thrift.protocol.TCompactProtocol(new org.apache.thrift.transport.TIOStreamTransport(out)));
+ } catch (org.apache.thrift.TException te) {
+ throw new java.io.IOException(te);
+ }
+ }
+
+ private void readObject(java.io.ObjectInputStream in) throws java.io.IOException, java.lang.ClassNotFoundException {
+ try {
+ read(new org.apache.thrift.protocol.TCompactProtocol(new org.apache.thrift.transport.TIOStreamTransport(in)));
+ } catch (org.apache.thrift.TException te) {
+ throw new java.io.IOException(te);
+ }
+ }
+
+ private static class unRegisterInterpreterProcess_resultStandardSchemeFactory implements org.apache.thrift.scheme.SchemeFactory {
+ public unRegisterInterpreterProcess_resultStandardScheme getScheme() {
+ return new unRegisterInterpreterProcess_resultStandardScheme();
+ }
+ }
+
+ private static class unRegisterInterpreterProcess_resultStandardScheme extends org.apache.thrift.scheme.StandardScheme<unRegisterInterpreterProcess_result> {
+
+ public void read(org.apache.thrift.protocol.TProtocol iprot, unRegisterInterpreterProcess_result struct) throws org.apache.thrift.TException {
+ org.apache.thrift.protocol.TField schemeField;
+ iprot.readStructBegin();
+ while (true)
+ {
+ schemeField = iprot.readFieldBegin();
+ if (schemeField.type == org.apache.thrift.protocol.TType.STOP) {
+ break;
+ }
+ switch (schemeField.id) {
+ default:
+ org.apache.thrift.protocol.TProtocolUtil.skip(iprot, schemeField.type);
+ }
+ iprot.readFieldEnd();
+ }
+ iprot.readStructEnd();
+
+ // check for required fields of primitive type, which can't be checked in the validate method
+ struct.validate();
+ }
+
+ public void write(org.apache.thrift.protocol.TProtocol oprot, unRegisterInterpreterProcess_result struct) throws org.apache.thrift.TException {
+ struct.validate();
+
+ oprot.writeStructBegin(STRUCT_DESC);
+ oprot.writeFieldStop();
+ oprot.writeStructEnd();
+ }
+
+ }
+
+ private static class unRegisterInterpreterProcess_resultTupleSchemeFactory implements org.apache.thrift.scheme.SchemeFactory {
+ public unRegisterInterpreterProcess_resultTupleScheme getScheme() {
+ return new unRegisterInterpreterProcess_resultTupleScheme();
+ }
+ }
+
+ private static class unRegisterInterpreterProcess_resultTupleScheme extends org.apache.thrift.scheme.TupleScheme<unRegisterInterpreterProcess_result> {
+
+ @Override
+ public void write(org.apache.thrift.protocol.TProtocol prot, unRegisterInterpreterProcess_result struct) throws org.apache.thrift.TException {
+ org.apache.thrift.protocol.TTupleProtocol oprot = (org.apache.thrift.protocol.TTupleProtocol) prot;
+ }
+
+ @Override
+ public void read(org.apache.thrift.protocol.TProtocol prot, unRegisterInterpreterProcess_result struct) throws org.apache.thrift.TException {
+ org.apache.thrift.protocol.TTupleProtocol iprot = (org.apache.thrift.protocol.TTupleProtocol) prot;
+ }
+ }
+
+ private static <S extends org.apache.thrift.scheme.IScheme> S scheme(org.apache.thrift.protocol.TProtocol proto) {
+ return (org.apache.thrift.scheme.StandardScheme.class.equals(proto.getScheme()) ? STANDARD_SCHEME_FACTORY : TUPLE_SCHEME_FACTORY).getScheme();
+ }
+ }
+
public static class appendOutput_args implements org.apache.thrift.TBase<appendOutput_args, appendOutput_args._Fields>, java.io.Serializable, Cloneable, Comparable<appendOutput_args> {
private static final org.apache.thrift.protocol.TStruct STRUCT_DESC = new org.apache.thrift.protocol.TStruct("appendOutput_args");
diff --git a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/RemoteInterpreterEventType.java b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/RemoteInterpreterEventType.java
index 892bd06..7007b48 100644
--- a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/RemoteInterpreterEventType.java
+++ b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/RemoteInterpreterEventType.java
@@ -24,7 +24,7 @@
package org.apache.zeppelin.interpreter.thrift;
-@javax.annotation.Generated(value = "Autogenerated by Thrift Compiler (0.13.0)", date = "2020-07-28")
+@javax.annotation.Generated(value = "Autogenerated by Thrift Compiler (0.13.0)", date = "2020-09-14")
public enum RemoteInterpreterEventType implements org.apache.thrift.TEnum {
NO_OP(1),
ANGULAR_OBJECT_ADD(2),
diff --git a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/RemoteInterpreterResult.java b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/RemoteInterpreterResult.java
index fb8d25e..5f92206 100644
--- a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/RemoteInterpreterResult.java
+++ b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/RemoteInterpreterResult.java
@@ -24,7 +24,7 @@
package org.apache.zeppelin.interpreter.thrift;
@SuppressWarnings({"cast", "rawtypes", "serial", "unchecked", "unused"})
-@javax.annotation.Generated(value = "Autogenerated by Thrift Compiler (0.13.0)", date = "2020-07-28")
+@javax.annotation.Generated(value = "Autogenerated by Thrift Compiler (0.13.0)", date = "2020-09-14")
public class RemoteInterpreterResult implements org.apache.thrift.TBase<RemoteInterpreterResult, RemoteInterpreterResult._Fields>, java.io.Serializable, Cloneable, Comparable<RemoteInterpreterResult> {
private static final org.apache.thrift.protocol.TStruct STRUCT_DESC = new org.apache.thrift.protocol.TStruct("RemoteInterpreterResult");
diff --git a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/RemoteInterpreterResultMessage.java b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/RemoteInterpreterResultMessage.java
index e9fca08..fb9f60f 100644
--- a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/RemoteInterpreterResultMessage.java
+++ b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/RemoteInterpreterResultMessage.java
@@ -24,7 +24,7 @@
package org.apache.zeppelin.interpreter.thrift;
@SuppressWarnings({"cast", "rawtypes", "serial", "unchecked", "unused"})
-@javax.annotation.Generated(value = "Autogenerated by Thrift Compiler (0.13.0)", date = "2020-07-28")
+@javax.annotation.Generated(value = "Autogenerated by Thrift Compiler (0.13.0)", date = "2020-09-14")
public class RemoteInterpreterResultMessage implements org.apache.thrift.TBase<RemoteInterpreterResultMessage, RemoteInterpreterResultMessage._Fields>, java.io.Serializable, Cloneable, Comparable<RemoteInterpreterResultMessage> {
private static final org.apache.thrift.protocol.TStruct STRUCT_DESC = new org.apache.thrift.protocol.TStruct("RemoteInterpreterResultMessage");
diff --git a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/RemoteInterpreterService.java b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/RemoteInterpreterService.java
index 9f80238..f4d150b 100644
--- a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/RemoteInterpreterService.java
+++ b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/RemoteInterpreterService.java
@@ -24,7 +24,7 @@
package org.apache.zeppelin.interpreter.thrift;
@SuppressWarnings({"cast", "rawtypes", "serial", "unchecked", "unused"})
-@javax.annotation.Generated(value = "Autogenerated by Thrift Compiler (0.13.0)", date = "2020-07-28")
+@javax.annotation.Generated(value = "Autogenerated by Thrift Compiler (0.13.0)", date = "2020-09-14")
public class RemoteInterpreterService {
public interface Iface {
diff --git a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/RunParagraphsEvent.java b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/RunParagraphsEvent.java
index e424597..f975521 100644
--- a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/RunParagraphsEvent.java
+++ b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/RunParagraphsEvent.java
@@ -24,7 +24,7 @@
package org.apache.zeppelin.interpreter.thrift;
@SuppressWarnings({"cast", "rawtypes", "serial", "unchecked", "unused"})
-@javax.annotation.Generated(value = "Autogenerated by Thrift Compiler (0.13.0)", date = "2020-07-28")
+@javax.annotation.Generated(value = "Autogenerated by Thrift Compiler (0.13.0)", date = "2020-09-14")
public class RunParagraphsEvent implements org.apache.thrift.TBase<RunParagraphsEvent, RunParagraphsEvent._Fields>, java.io.Serializable, Cloneable, Comparable<RunParagraphsEvent> {
private static final org.apache.thrift.protocol.TStruct STRUCT_DESC = new org.apache.thrift.protocol.TStruct("RunParagraphsEvent");
diff --git a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/ServiceException.java b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/ServiceException.java
index 90575c6..931cdd6 100644
--- a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/ServiceException.java
+++ b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/ServiceException.java
@@ -24,7 +24,7 @@
package org.apache.zeppelin.interpreter.thrift;
@SuppressWarnings({"cast", "rawtypes", "serial", "unchecked", "unused"})
-@javax.annotation.Generated(value = "Autogenerated by Thrift Compiler (0.13.0)", date = "2020-07-28")
+@javax.annotation.Generated(value = "Autogenerated by Thrift Compiler (0.13.0)", date = "2020-09-14")
public class ServiceException extends org.apache.thrift.TException implements org.apache.thrift.TBase<ServiceException, ServiceException._Fields>, java.io.Serializable, Cloneable, Comparable<ServiceException> {
private static final org.apache.thrift.protocol.TStruct STRUCT_DESC = new org.apache.thrift.protocol.TStruct("ServiceException");
diff --git a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/WebUrlInfo.java b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/WebUrlInfo.java
index f5a9e24..df1fad8 100644
--- a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/WebUrlInfo.java
+++ b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/thrift/WebUrlInfo.java
@@ -24,7 +24,7 @@
package org.apache.zeppelin.interpreter.thrift;
@SuppressWarnings({"cast", "rawtypes", "serial", "unchecked", "unused"})
-@javax.annotation.Generated(value = "Autogenerated by Thrift Compiler (0.13.0)", date = "2020-07-28")
+@javax.annotation.Generated(value = "Autogenerated by Thrift Compiler (0.13.0)", date = "2020-09-14")
public class WebUrlInfo implements org.apache.thrift.TBase<WebUrlInfo, WebUrlInfo._Fields>, java.io.Serializable, Cloneable, Comparable<WebUrlInfo> {
private static final org.apache.thrift.protocol.TStruct STRUCT_DESC = new org.apache.thrift.protocol.TStruct("WebUrlInfo");
diff --git a/zeppelin-interpreter/src/main/thrift/RemoteInterpreterEventService.thrift b/zeppelin-interpreter/src/main/thrift/RemoteInterpreterEventService.thrift
index 479ee94..de695cb 100644
--- a/zeppelin-interpreter/src/main/thrift/RemoteInterpreterEventService.thrift
+++ b/zeppelin-interpreter/src/main/thrift/RemoteInterpreterEventService.thrift
@@ -104,6 +104,7 @@ exception ServiceException{
service RemoteInterpreterEventService {
void registerInterpreterProcess(1: RegisterInfo registerInfo);
+ void unRegisterInterpreterProcess(1: string intpGroupId);
void appendOutput(1: OutputAppendEvent event);
void updateOutput(1: OutputUpdateEvent event);
diff --git a/zeppelin-interpreter/src/test/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreterServerTest.java b/zeppelin-interpreter/src/test/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreterServerTest.java
index ebbeea1..0e9a42a 100644
--- a/zeppelin-interpreter/src/test/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreterServerTest.java
+++ b/zeppelin-interpreter/src/test/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreterServerTest.java
@@ -41,7 +41,7 @@ import static org.mockito.Mockito.mock;
public class RemoteInterpreterServerTest {
@Test
- public void testStartStop() throws InterruptedException, IOException, TException {
+ public void testStartStop() throws Exception {
RemoteInterpreterServer server = new RemoteInterpreterServer("localhost",
RemoteInterpreterUtils.findRandomAvailablePortOnAllLocalInterfaces(), ":", "groupId", true);
@@ -50,7 +50,7 @@ public class RemoteInterpreterServerTest {
}
@Test
- public void testStartStopWithQueuedEvents() throws InterruptedException, IOException, TException {
+ public void testStartStopWithQueuedEvents() throws Exception {
RemoteInterpreterServer server = new RemoteInterpreterServer("localhost",
RemoteInterpreterUtils.findRandomAvailablePortOnAllLocalInterfaces(), ":", "groupId", true);
server.intpEventClient = mock(RemoteInterpreterEventClient.class);
@@ -93,7 +93,7 @@ public class RemoteInterpreterServerTest {
}
@Test
- public void testInterpreter() throws IOException, TException, InterruptedException {
+ public void testInterpreter() throws Exception {
final RemoteInterpreterServer server = new RemoteInterpreterServer("localhost",
RemoteInterpreterUtils.findRandomAvailablePortOnAllLocalInterfaces(), ":", "groupId", true);
server.intpEventClient = mock(RemoteInterpreterEventClient.class);
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 12d1c2b..5417ad7 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
@@ -40,7 +40,6 @@ import org.apache.zeppelin.display.AngularObjectRegistryListener;
import org.apache.zeppelin.helium.ApplicationEventListener;
import org.apache.zeppelin.interpreter.launcher.InterpreterLaunchContext;
import org.apache.zeppelin.interpreter.launcher.InterpreterLauncher;
-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;
@@ -137,7 +136,6 @@ public class InterpreterSetting {
private transient ZeppelinConfiguration conf = new ZeppelinConfiguration();
- private transient LifecycleManager lifecycleManager;
private transient RecoveryStorage recoveryStorage;
private transient RemoteInterpreterEventServer interpreterEventServer;
@@ -243,11 +241,6 @@ public class InterpreterSetting {
return this;
}
- public Builder setLifecycleManager(LifecycleManager lifecycleManager) {
- interpreterSetting.lifecycleManager = lifecycleManager;
- return this;
- }
-
public Builder setRecoveryStorage(RecoveryStorage recoveryStorage) {
interpreterSetting.recoveryStorage = recoveryStorage;
return this;
@@ -269,9 +262,6 @@ public class InterpreterSetting {
void postProcessing() {
this.id = this.name;
- if (this.lifecycleManager == null) {
- this.lifecycleManager = new NullLifecycleManager(conf);
- }
if (this.recoveryStorage == null) {
try {
this.recoveryStorage = new NullRecoveryStorage(conf, interpreterSettingManager);
@@ -354,11 +344,6 @@ public class InterpreterSetting {
return this;
}
- public InterpreterSetting setLifecycleManager(LifecycleManager lifecycleManager) {
- this.lifecycleManager = lifecycleManager;
- return this;
- }
-
public InterpreterSetting setRecoveryStorage(RecoveryStorage recoveryStorage) {
this.recoveryStorage = recoveryStorage;
return this;
@@ -397,10 +382,6 @@ public class InterpreterSetting {
return recoveryStorage;
}
- public LifecycleManager getLifecycleManager() {
- return lifecycleManager;
- }
-
public String getId() {
return id;
}
@@ -844,7 +825,7 @@ public class InterpreterSetting {
Properties intpProperties = getJavaProperties();
for (InterpreterInfo info : interpreterInfos) {
Interpreter interpreter = new RemoteInterpreter(intpProperties, sessionId,
- info.getClassName(), user, lifecycleManager);
+ info.getClassName(), user);
if (info.isDefaultInterpreter()) {
interpreters.add(0, interpreter);
} else {
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 ed02c0a..377b525 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
@@ -136,7 +136,6 @@ public class InterpreterSettingManager implements NoteEventListener, ClusterEven
private RemoteInterpreterProcessListener remoteInterpreterProcessListener;
private ApplicationEventListener appEventListener;
private DependencyResolver dependencyResolver;
- private LifecycleManager lifecycleManager;
private RecoveryStorage recoveryStorage;
private ConfigStorage configStorage;
private RemoteInterpreterEventServer interpreterEventServer;
@@ -187,13 +186,8 @@ public class InterpreterSettingManager implements NoteEventListener, ClusterEven
conf.getRecoveryStorageClass(),
new Class[] {ZeppelinConfiguration.class, InterpreterSettingManager.class},
new Object[] {conf, this});
- 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());
+
+ LOGGER.info("Using RecoveryStorage: " + this.recoveryStorage.getClass().getName());
this.configStorage = configStorage;
init();
@@ -227,7 +221,6 @@ public class InterpreterSettingManager implements NoteEventListener, ClusterEven
.setRemoteInterpreterProcessListener(remoteInterpreterProcessListener)
.setAppEventListener(appEventListener)
.setDependencyResolver(dependencyResolver)
- .setLifecycleManager(lifecycleManager)
.setRecoveryStorage(recoveryStorage)
.setInterpreterEventServer(interpreterEventServer)
.postProcessing();
@@ -641,6 +634,13 @@ public class InterpreterSettingManager implements NoteEventListener, ClusterEven
return interpreterGroups;
}
+ // TODO(zjffdu) Current approach is not optimized. we have to iterate all interpreter settings.
+ public void removeInterpreterGroup(String intpGroupId) {
+ for (InterpreterSetting interpreterSetting : interpreterSettings.values()) {
+ interpreterSetting.removeInterpreterGroup(intpGroupId);
+ }
+ }
+
//TODO(zjffdu) move Resource related api to ResourceManager
public ResourceSet getAllResources() {
return getAllResourcesExcept(null);
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 a1a478e..1d13fef 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
@@ -65,7 +65,6 @@ public class ManagedInterpreterGroup extends InterpreterGroup {
remoteInterpreterProcess = interpreterSetting.createInterpreterProcess(id, userName,
properties);
remoteInterpreterProcess.start(userName);
- interpreterSetting.getLifecycleManager().onInterpreterProcessStarted(this);
getInterpreterSetting().getRecoveryStorage()
.onInterpreterClientStart(remoteInterpreterProcess);
}
diff --git a/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/RemoteInterpreterEventServer.java b/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/RemoteInterpreterEventServer.java
index 819f02b..a079c47 100644
--- a/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/RemoteInterpreterEventServer.java
+++ b/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/RemoteInterpreterEventServer.java
@@ -160,26 +160,42 @@ public class RemoteInterpreterEventServer implements RemoteInterpreterEventServi
InterpreterGroup interpreterGroup =
interpreterSettingManager.getInterpreterGroupById(registerInfo.getInterpreterGroupId());
if (interpreterGroup == null) {
- LOGGER.warn("No such interpreterGroup: " + registerInfo.getInterpreterGroupId());
+ LOGGER.warn("Unable to register interpreter process, because no such interpreterGroup: {}",
+ registerInfo.getInterpreterGroupId());
return;
}
RemoteInterpreterProcess interpreterProcess =
((ManagedInterpreterGroup) interpreterGroup).getInterpreterProcess();
if (interpreterProcess == null) {
- LOGGER.warn("Interpreter process does not existed yet for InterpreterGroup: " +
- registerInfo.getInterpreterGroupId());
+ LOGGER.warn("Unable to register interpreter process, because no interpreter process associated with " +
+ "interpreterGroup: {}", registerInfo.getInterpreterGroupId());
+ return;
}
- LOGGER.info("Register interpreter process: {}:{}, {}",
+ LOGGER.info("Register interpreter process: {}:{}, interpreterGroup: {}",
registerInfo.getHost(), registerInfo.getPort(), registerInfo.getInterpreterGroupId());
interpreterProcess.processStarted(registerInfo.port, registerInfo.host);
}
@Override
+ public void unRegisterInterpreterProcess(String intpGroupId) throws TException {
+ LOGGER.info("Unregister interpreter process: {}", intpGroupId);
+ InterpreterGroup interpreterGroup =
+ interpreterSettingManager.getInterpreterGroupById(intpGroupId);
+ if (interpreterGroup == null) {
+ LOGGER.warn("Unable to unregister interpreter process because no such interpreterGroup: {}",
+ intpGroupId);
+ return;
+ }
+ interpreterSettingManager.removeInterpreterGroup(intpGroupId);
+ }
+
+ @Override
public void sendWebUrl(WebUrlInfo weburlInfo) throws TException {
InterpreterGroup interpreterGroup =
interpreterSettingManager.getInterpreterGroupById(weburlInfo.getInterpreterGroupId());
if (interpreterGroup == null) {
- LOGGER.warn("No such interpreterGroup: " + weburlInfo.getInterpreterGroupId());
+ LOGGER.warn("Unable to sendWebUrl, because no such interpreterGroup: {}",
+ weburlInfo.getInterpreterGroupId());
return;
}
interpreterGroup.setWebUrl(weburlInfo.getWeburl());
diff --git a/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/lifecycle/TimeoutLifecycleManager.java b/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/lifecycle/TimeoutLifecycleManager.java
deleted file mode 100644
index 390159b..0000000
--- a/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/lifecycle/TimeoutLifecycleManager.java
+++ /dev/null
@@ -1,74 +0,0 @@
-package org.apache.zeppelin.interpreter.lifecycle;
-
-import org.apache.zeppelin.conf.ZeppelinConfiguration;
-import org.apache.zeppelin.interpreter.LifecycleManager;
-import org.apache.zeppelin.interpreter.ManagedInterpreterGroup;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.util.Map;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.Executors;
-import java.util.concurrent.ScheduledExecutorService;
-
-import static java.util.concurrent.TimeUnit.MILLISECONDS;
-
-
-/**
- * This lifecycle manager would close interpreter after it is timeout. By default, it is timeout
- * after no using in 1 hour.
- *
- * For now, this class only manage the lifecycle of interpreter group (will close interpreter
- * process after timeout). Managing the lifecycle of interpreter session could be done in future
- * if necessary.
- */
-public class TimeoutLifecycleManager implements LifecycleManager {
-
- private static final Logger LOGGER = LoggerFactory.getLogger(TimeoutLifecycleManager.class);
-
- // ManagerInterpreterGroup -> LastTimeUsing timestamp
- private Map<ManagedInterpreterGroup, Long> interpreterGroups = new ConcurrentHashMap<>();
-
- private long checkInterval;
- private long timeoutThreshold;
-
- private ScheduledExecutorService checkScheduler;
-
- public TimeoutLifecycleManager(ZeppelinConfiguration zConf) {
- this.checkInterval = zConf.getLong(ZeppelinConfiguration.ConfVars
- .ZEPPELIN_INTERPRETER_LIFECYCLE_MANAGER_TIMEOUT_CHECK_INTERVAL);
- this.timeoutThreshold = zConf.getLong(
- ZeppelinConfiguration.ConfVars.ZEPPELIN_INTERPRETER_LIFECYCLE_MANAGER_TIMEOUT_THRESHOLD);
- this.checkScheduler = Executors.newScheduledThreadPool(1);
- this.checkScheduler.scheduleAtFixedRate(() -> {
- long now = System.currentTimeMillis();
- for (Map.Entry<ManagedInterpreterGroup, Long> entry : interpreterGroups.entrySet()) {
- ManagedInterpreterGroup interpreterGroup = entry.getKey();
- Long lastTimeUsing = entry.getValue();
- if ((now - lastTimeUsing) > timeoutThreshold) {
- LOGGER.info("InterpreterGroup {} is timeout.", interpreterGroup.getId());
- try {
- interpreterGroup.close();
- } catch (Exception e) {
- LOGGER.warn("Fail to close interpreterGroup: " + interpreterGroup.getId(), e);
- }
- interpreterGroups.remove(entry.getKey());
- }
- }
- }, checkInterval, checkInterval, MILLISECONDS);
- LOGGER.info("TimeoutLifecycleManager is started with checkinterval: " + checkInterval
- + ", timeoutThreshold: " + timeoutThreshold);
- }
-
- @Override
- public void onInterpreterProcessStarted(ManagedInterpreterGroup interpreterGroup) {
- LOGGER.info("Process of InterpreterGroup {} is started", interpreterGroup.getId());
- interpreterGroups.put(interpreterGroup, System.currentTimeMillis());
- }
-
- @Override
- public void onInterpreterUse(ManagedInterpreterGroup interpreterGroup, String sessionId) {
- LOGGER.debug("InterpreterGroup {} is used in session {}", interpreterGroup.getId(), sessionId);
- interpreterGroups.put(interpreterGroup, System.currentTimeMillis());
- }
-}
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 7f323e0..754c4ae 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
@@ -65,21 +65,17 @@ public class RemoteInterpreter extends Interpreter {
private volatile boolean isOpened = false;
private volatile boolean isCreated = false;
- private LifecycleManager lifecycleManager;
-
/**
* Remote interpreter and manage interpreter process
*/
public RemoteInterpreter(Properties properties,
String sessionId,
String className,
- String userName,
- LifecycleManager lifecycleManager) {
+ String userName) {
super(properties);
this.sessionId = sessionId;
this.className = className;
this.setUserName(userName);
- this.lifecycleManager = lifecycleManager;
}
public boolean isOpened() {
@@ -147,7 +143,6 @@ public class RemoteInterpreter extends Interpreter {
return null;
});
isOpened = true;
- this.lifecycleManager.onInterpreterUse(this.getInterpreterGroup(), sessionId);
}
}
}
@@ -186,7 +181,6 @@ public class RemoteInterpreter extends Interpreter {
return null;
});
isOpened = false;
- this.lifecycleManager.onInterpreterUse(this.getInterpreterGroup(), sessionId);
} else {
LOGGER.warn("close is called when RemoterInterpreter is not opened for " + className);
}
@@ -210,7 +204,6 @@ public class RemoteInterpreter extends Interpreter {
return new InterpreterResult(InterpreterResult.Code.ERROR,
"Interpreter process is not running\n" + interpreterProcess.getErrorMessage());
}
- this.lifecycleManager.onInterpreterUse(this.getInterpreterGroup(), sessionId);
return interpreterProcess.callRemoteFunction(client -> {
RemoteInterpreterResult remoteResult = client.interpret(
sessionId, className, st, convert(context));
@@ -260,7 +253,6 @@ public class RemoteInterpreter extends Interpreter {
} catch (IOException e) {
throw new InterpreterException(e);
}
- this.lifecycleManager.onInterpreterUse(this.getInterpreterGroup(), sessionId);
interpreterProcess.callRemoteFunction(client -> {
client.cancel(sessionId, className, convert(context));
return null;
@@ -286,7 +278,6 @@ public class RemoteInterpreter extends Interpreter {
throw new InterpreterException(e);
}
- this.lifecycleManager.onInterpreterUse(this.getInterpreterGroup(), sessionId);
FormType type = interpreterProcess.callRemoteFunction(client -> {
formType = FormType.valueOf(client.getFormType(sessionId, className));
return formType;
@@ -307,7 +298,6 @@ public class RemoteInterpreter extends Interpreter {
} catch (IOException e) {
throw new InterpreterException(e);
}
- this.lifecycleManager.onInterpreterUse(this.getInterpreterGroup(), sessionId);
return interpreterProcess.callRemoteFunction(client ->
client.getProgress(sessionId, className, convert(context)));
}
@@ -326,7 +316,6 @@ public class RemoteInterpreter extends Interpreter {
} catch (IOException e) {
throw new InterpreterException(e);
}
- this.lifecycleManager.onInterpreterUse(this.getInterpreterGroup(), sessionId);
return interpreterProcess.callRemoteFunction(client ->
client.completion(sessionId, className, buf, cursor, convert(interpreterContext)));
}
@@ -342,7 +331,6 @@ public class RemoteInterpreter extends Interpreter {
} catch (IOException e) {
throw new RuntimeException(e);
}
- this.lifecycleManager.onInterpreterUse(this.getInterpreterGroup(), sessionId);
return interpreterProcess.callRemoteFunction(client -> {
return client.getStatus(sessionId, jobId);
});
diff --git a/zeppelin-zengine/src/test/java/org/apache/zeppelin/interpreter/InterpreterSettingManagerTest.java b/zeppelin-zengine/src/test/java/org/apache/zeppelin/interpreter/InterpreterSettingManagerTest.java
index d8a61a3..6e8f0be 100644
--- a/zeppelin-zengine/src/test/java/org/apache/zeppelin/interpreter/InterpreterSettingManagerTest.java
+++ b/zeppelin-zengine/src/test/java/org/apache/zeppelin/interpreter/InterpreterSettingManagerTest.java
@@ -22,7 +22,6 @@ import org.apache.zeppelin.conf.ZeppelinConfiguration;
import org.apache.zeppelin.dep.Dependency;
import org.apache.zeppelin.display.AngularObjectRegistryListener;
import org.apache.zeppelin.helium.ApplicationEventListener;
-import org.apache.zeppelin.interpreter.lifecycle.NullLifecycleManager;
import org.apache.zeppelin.interpreter.remote.RemoteInterpreterProcessListener;
import org.junit.Test;
import org.sonatype.aether.RepositoryException;
@@ -50,7 +49,6 @@ public class InterpreterSettingManagerTest extends AbstractInterpreterTest {
InterpreterSetting interpreterSetting = interpreterSettingManager.getByName("test");
assertEquals("test", interpreterSetting.getName());
assertEquals("test", interpreterSetting.getGroup());
- assertTrue(interpreterSetting.getLifecycleManager() instanceof NullLifecycleManager);
assertEquals(8, interpreterSetting.getInterpreterInfos().size());
// 3 other builtin properties:
// * zeppelin.interpreter.output.limit
diff --git a/zeppelin-zengine/src/test/java/org/apache/zeppelin/interpreter/SessionConfInterpreterTest.java b/zeppelin-zengine/src/test/java/org/apache/zeppelin/interpreter/SessionConfInterpreterTest.java
index 7baa2e2..e0f54f0 100644
--- a/zeppelin-zengine/src/test/java/org/apache/zeppelin/interpreter/SessionConfInterpreterTest.java
+++ b/zeppelin-zengine/src/test/java/org/apache/zeppelin/interpreter/SessionConfInterpreterTest.java
@@ -45,7 +45,7 @@ public class SessionConfInterpreterTest {
properties, "session_1", "group_1", mockInterpreterSetting);
RemoteInterpreter remoteInterpreter =
- new RemoteInterpreter(properties, "session_1", "clasName", "user1", null);
+ new RemoteInterpreter(properties, "session_1", "clasName", "user1");
List<Interpreter> interpreters = new ArrayList<>();
interpreters.add(confInterpreter);
interpreters.add(remoteInterpreter);
diff --git a/zeppelin-zengine/src/test/java/org/apache/zeppelin/interpreter/lifecycle/TimeoutLifecycleManagerTest.java b/zeppelin-zengine/src/test/java/org/apache/zeppelin/interpreter/lifecycle/TimeoutLifecycleManagerTest.java
index a0cdc29..20b6d61 100644
--- a/zeppelin-zengine/src/test/java/org/apache/zeppelin/interpreter/lifecycle/TimeoutLifecycleManagerTest.java
+++ b/zeppelin-zengine/src/test/java/org/apache/zeppelin/interpreter/lifecycle/TimeoutLifecycleManagerTest.java
@@ -27,6 +27,8 @@ import org.apache.zeppelin.interpreter.remote.RemoteInterpreter;
import org.apache.zeppelin.scheduler.Job;
import org.junit.Test;
+import java.io.File;
+import java.io.FileWriter;
import java.io.IOException;
import java.util.Map;
@@ -36,15 +38,25 @@ import static org.junit.Assert.assertTrue;
public class TimeoutLifecycleManagerTest extends AbstractInterpreterTest {
+ private File zeppelinSiteFile = new File("zeppelin-site.xml");
+
@Override
public void setUp() throws Exception {
- System.setProperty(ZeppelinConfiguration.ConfVars.ZEPPELIN_INTERPRETER_LIFECYCLE_MANAGER_CLASS.getVarName(),
+ ZeppelinConfiguration zConf = ZeppelinConfiguration.create();
+ zConf.setProperty(ZeppelinConfiguration.ConfVars.ZEPPELIN_INTERPRETER_LIFECYCLE_MANAGER_CLASS.getVarName(),
TimeoutLifecycleManager.class.getName());
- System.setProperty(ZeppelinConfiguration.ConfVars.ZEPPELIN_INTERPRETER_LIFECYCLE_MANAGER_TIMEOUT_CHECK_INTERVAL.getVarName(), "1000");
- System.setProperty(ZeppelinConfiguration.ConfVars.ZEPPELIN_INTERPRETER_LIFECYCLE_MANAGER_TIMEOUT_THRESHOLD.getVarName(), "10000");
+ zConf.setProperty(ZeppelinConfiguration.ConfVars.ZEPPELIN_INTERPRETER_LIFECYCLE_MANAGER_TIMEOUT_CHECK_INTERVAL.getVarName(), "1000");
+ zConf.setProperty(ZeppelinConfiguration.ConfVars.ZEPPELIN_INTERPRETER_LIFECYCLE_MANAGER_TIMEOUT_THRESHOLD.getVarName(), "10000");
+
+ zConf.save(zeppelinSiteFile);
super.setUp();
}
+ @Override
+ public void tearDown() {
+ zeppelinSiteFile.delete();
+ }
+
@Test
public void testTimeout_1() throws InterpreterException, InterruptedException, IOException {
assertTrue(interpreterFactory.getInterpreter("test.echo", new ExecutionContextBuilder().setUser("user1").setNoteId("note1").setDefaultInterpreterGroup("test").createExecutionContext()) instanceof RemoteInterpreter);
@@ -66,7 +78,6 @@ public class TimeoutLifecycleManagerTest extends AbstractInterpreterTest {
Thread.sleep(15 * 1000);
// interpreterGroup is timeout, so is removed.
assertEquals(0, interpreterSetting.getAllInterpreterGroups().size());
- assertFalse(remoteInterpreter.isOpened());
}
@Test