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 2019/03/30 13:18:28 UTC
[zeppelin] branch master updated (0b8423c -> 06f09cf)
This is an automated email from the ASF dual-hosted git repository.
zjffdu pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/zeppelin.git.
from 0b8423c [ZEPPELIN-3985]. Move note permission from notebook-authorization.json to note file
new 6c24d1f Revert "[ZEPPELIN-4089] handle ipython kernel crash"
new 06f09cf Revert "ZEPPELIN-4081. when the python process is killed,the task state is still running"
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
LICENSE | 1 -
python/pom.xml | 8 ---
.../org/apache/zeppelin/python/IPythonClient.java | 7 +--
.../apache/zeppelin/python/IPythonInterpreter.java | 49 ++++-----------
.../apache/zeppelin/python/PythonInterpreter.java | 23 ++-----
.../main/resources/grpc/python/ipython_server.py | 44 ++++---------
.../zeppelin/python/BasePythonInterpreterTest.java | 3 +-
.../zeppelin/python/IPythonInterpreterTest.java | 73 ----------------------
.../zeppelin/python/PythonInterpreterTest.java | 33 +---------
spark/interpreter/pom.xml | 6 --
.../apache/zeppelin/spark/IPySparkInterpreter.java | 3 +-
11 files changed, 33 insertions(+), 217 deletions(-)
[zeppelin] 02/02: Revert "ZEPPELIN-4081. when the python process is
killed, the task state is still running"
Posted by zj...@apache.org.
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
commit 06f09cf3ebba677f2115ba4cca554c8860cff0a9
Author: Jeff Zhang <zj...@apache.org>
AuthorDate: Fri Mar 29 22:26:50 2019 +0800
Revert "ZEPPELIN-4081. when the python process is killed,the task state is still running"
This reverts commit c9514e26a00b5c2b76c358425bae79fee183c9e1.
---
LICENSE | 1 -
python/pom.xml | 8 ----
.../org/apache/zeppelin/python/IPythonClient.java | 7 +---
.../apache/zeppelin/python/IPythonInterpreter.java | 49 ++++++----------------
.../apache/zeppelin/python/PythonInterpreter.java | 23 ++--------
.../zeppelin/python/BasePythonInterpreterTest.java | 3 +-
.../zeppelin/python/IPythonInterpreterTest.java | 29 -------------
.../zeppelin/python/PythonInterpreterTest.java | 33 +--------------
spark/interpreter/pom.xml | 6 ---
.../apache/zeppelin/spark/IPySparkInterpreter.java | 3 +-
10 files changed, 21 insertions(+), 141 deletions(-)
diff --git a/LICENSE b/LICENSE
index a456a41..3b34053 100644
--- a/LICENSE
+++ b/LICENSE
@@ -260,7 +260,6 @@ The text of each license is also included at licenses/LICENSE-[project]-[version
(Apache 2.0) Nimbus JOSE+JWT (https://bitbucket.org/connect2id/nimbus-jose-jwt/wiki/Home)
(Apache 2.0) jarchivelib (https://github.com/thrau/jarchivelib)
(Apache 2.0) Google Cloud Client Library for Java (https://github.com/GoogleCloudPlatform/google-cloud-java)
- (Apache 2.0) concurrentunit (https://github.com/jhalterman/concurrentunit)
========================================================================
BSD 3-Clause licenses
diff --git a/python/pom.xml b/python/pom.xml
index 2700cba..d9371ed 100644
--- a/python/pom.xml
+++ b/python/pom.xml
@@ -87,14 +87,6 @@
<artifactId>mockito-all</artifactId>
<scope>test</scope>
</dependency>
-
- <dependency>
- <groupId>net.jodah</groupId>
- <artifactId>concurrentunit</artifactId>
- <version>0.4.4</version>
- <scope>test</scope>
- </dependency>
-
</dependencies>
<build>
diff --git a/python/src/main/java/org/apache/zeppelin/python/IPythonClient.java b/python/src/main/java/org/apache/zeppelin/python/IPythonClient.java
index c729898..b9c897b 100644
--- a/python/src/main/java/org/apache/zeppelin/python/IPythonClient.java
+++ b/python/src/main/java/org/apache/zeppelin/python/IPythonClient.java
@@ -53,7 +53,6 @@ public class IPythonClient {
private final ManagedChannel channel;
private final IPythonGrpc.IPythonBlockingStub blockingStub;
private final IPythonGrpc.IPythonStub asyncStub;
- private volatile boolean maybeIPythonFailed = false;
private SecureRandom random = new SecureRandom();
@@ -84,7 +83,6 @@ public class IPythonClient {
final ExecuteResponse.Builder finalResponseBuilder = ExecuteResponse.newBuilder()
.setStatus(ExecuteStatus.SUCCESS);
final AtomicBoolean completedFlag = new AtomicBoolean(false);
- maybeIPythonFailed = false;
LOGGER.debug("stream_execute code:\n" + request.getCode());
asyncStub.execute(request, new StreamObserver<ExecuteResponse>() {
int index = 0;
@@ -139,7 +137,7 @@ public class IPythonClient {
}
LOGGER.error("Fail to call IPython grpc", throwable);
finalResponseBuilder.setStatus(ExecuteStatus.ERROR);
- maybeIPythonFailed = true;
+
completedFlag.set(true);
synchronized (completedFlag) {
completedFlag.notify();
@@ -206,9 +204,6 @@ public class IPythonClient {
asyncStub.stop(request, null);
}
- public boolean isMaybeIPythonFailed() {
- return maybeIPythonFailed;
- }
public static void main(String[] args) {
IPythonClient client = new IPythonClient("localhost", 50053);
diff --git a/python/src/main/java/org/apache/zeppelin/python/IPythonInterpreter.java b/python/src/main/java/org/apache/zeppelin/python/IPythonInterpreter.java
index f4c753d..9e23d04 100644
--- a/python/src/main/java/org/apache/zeppelin/python/IPythonInterpreter.java
+++ b/python/src/main/java/org/apache/zeppelin/python/IPythonInterpreter.java
@@ -82,7 +82,7 @@ public class IPythonInterpreter extends Interpreter implements ExecuteResultHand
private boolean useBuiltinPy4j = true;
private boolean usePy4JAuth = true;
private String secret;
- private volatile boolean pythonProcessRunning = false;
+ private volatile boolean pythonProcessFailed = false;
private InterpreterOutputStream interpreterOutput = new InterpreterOutputStream(LOGGER);
@@ -294,7 +294,7 @@ public class IPythonInterpreter extends Interpreter implements ExecuteResultHand
// wait until IPython kernel is started or timeout
long startTime = System.currentTimeMillis();
- while (!pythonProcessRunning) {
+ while (!pythonProcessFailed) {
try {
Thread.sleep(100);
} catch (InterruptedException e) {
@@ -305,7 +305,6 @@ public class IPythonInterpreter extends Interpreter implements ExecuteResultHand
StatusResponse response = ipythonClient.status(StatusRequest.newBuilder().build());
if (response.getStatus() == IPythonStatus.RUNNING) {
LOGGER.info("IPython Kernel is Running");
- pythonProcessRunning = true;
break;
} else {
LOGGER.info("Wait for IPython Kernel to be started");
@@ -320,7 +319,7 @@ public class IPythonInterpreter extends Interpreter implements ExecuteResultHand
+ " seconds");
}
}
- if (!pythonProcessRunning) {
+ if (pythonProcessFailed) {
throw new IOException("Fail to launch IPython Kernel as the python process is failed");
}
}
@@ -356,44 +355,23 @@ public class IPythonInterpreter extends Interpreter implements ExecuteResultHand
}
}
- public ExecuteWatchdog getWatchDog() {
- return watchDog;
- }
-
@Override
- public InterpreterResult interpret(String st,
- InterpreterContext context) throws InterpreterException {
+ public InterpreterResult interpret(String st, InterpreterContext context) {
zeppelinContext.setGui(context.getGui());
zeppelinContext.setNoteGui(context.getNoteGui());
zeppelinContext.setInterpreterContext(context);
interpreterOutput.setInterpreterOutput(context.out);
+ ExecuteResponse response =
+ ipythonClient.stream_execute(ExecuteRequest.newBuilder().setCode(st).build(),
+ interpreterOutput);
try {
- ExecuteResponse response =
- ipythonClient.stream_execute(ExecuteRequest.newBuilder().setCode(st).build(),
- interpreterOutput);
interpreterOutput.getInterpreterOutput().flush();
- // It is not known which method is called first (ipythonClient.stream_execute
- // or onProcessFailed) when ipython kernel process is exited. Because they are in
- // 2 different threads. So here we would check ipythonClient's status and sleep 1 second
- // if ipython kernel is maybe terminated.
- if (pythonProcessRunning && !ipythonClient.isMaybeIPythonFailed()) {
- return new InterpreterResult(
- InterpreterResult.Code.valueOf(response.getStatus().name()));
- } else {
- if (ipythonClient.isMaybeIPythonFailed()) {
- Thread.sleep(1000);
- }
- if (pythonProcessRunning) {
- return new InterpreterResult(
- InterpreterResult.Code.valueOf(response.getStatus().name()));
- } else {
- return new InterpreterResult(InterpreterResult.Code.ERROR,
- "IPython kernel is abnormally exited, please check your code and log.");
- }
- }
- } catch (Exception e) {
- throw new InterpreterException("Fail to interpret python code", e);
+ } catch (IOException e) {
+ throw new RuntimeException("Fail to write output", e);
}
+ InterpreterResult result = new InterpreterResult(
+ InterpreterResult.Code.valueOf(response.getStatus().name()));
+ return result;
}
@Override
@@ -438,13 +416,12 @@ public class IPythonInterpreter extends Interpreter implements ExecuteResultHand
@Override
public void onProcessComplete(int exitValue) {
LOGGER.warn("Python Process is completed with exitValue: " + exitValue);
- pythonProcessRunning = false;
}
@Override
public void onProcessFailed(ExecuteException e) {
LOGGER.warn("Exception happens in Python Process", e);
- pythonProcessRunning = false;
+ pythonProcessFailed = true;
}
static class ProcessLogOutputStream extends LogOutputStream {
diff --git a/python/src/main/java/org/apache/zeppelin/python/PythonInterpreter.java b/python/src/main/java/org/apache/zeppelin/python/PythonInterpreter.java
index c6770e5..fb4ba9c 100644
--- a/python/src/main/java/org/apache/zeppelin/python/PythonInterpreter.java
+++ b/python/src/main/java/org/apache/zeppelin/python/PythonInterpreter.java
@@ -17,7 +17,6 @@
package org.apache.zeppelin.python;
-import com.google.common.annotations.VisibleForTesting;
import com.google.common.io.Files;
import com.google.gson.Gson;
import org.apache.commons.exec.CommandLine;
@@ -161,10 +160,7 @@ public class PythonInterpreter extends Interpreter implements ExecuteResultHandl
pythonScriptRunning.set(true);
}
- @VisibleForTesting
- public DefaultExecutor getPythonExecutor() {
- return this.executor;
- }
+
private void createPythonScript() throws IOException {
// set java.io.tmpdir to /tmp on MacOS, because docker can not share the /var folder which will
@@ -352,7 +348,7 @@ public class PythonInterpreter extends Interpreter implements ExecuteResultHandl
}
synchronized (statementFinishedNotifier) {
- while (statementOutput == null && pythonScriptRunning.get()) {
+ while (statementOutput == null) {
try {
statementFinishedNotifier.wait(1000);
} catch (InterruptedException e) {
@@ -378,7 +374,7 @@ public class PythonInterpreter extends Interpreter implements ExecuteResultHandl
synchronized (pythonScriptInitialized) {
long startTime = System.currentTimeMillis();
- while (!pythonScriptInitialized.get() && pythonScriptRunning.get()
+ while (!pythonScriptInitialized.get()
&& System.currentTimeMillis() - startTime < MAX_TIMEOUT_SEC * 1000) {
try {
LOGGER.info("Wait for PythonScript initialized");
@@ -421,12 +417,7 @@ public class PythonInterpreter extends Interpreter implements ExecuteResultHandl
} catch (IOException e) {
throw new InterpreterException(e);
}
- if (pythonScriptRunning.get()) {
- return new InterpreterResult(Code.SUCCESS);
- } else {
- return new InterpreterResult(Code.ERROR,
- "Python process is abnormally exited, please check your code and log.");
- }
+ return new InterpreterResult(Code.SUCCESS);
}
}
@@ -599,9 +590,6 @@ public class PythonInterpreter extends Interpreter implements ExecuteResultHandl
LOGGER.info("python process terminated. exit code " + exitValue);
pythonScriptRunning.set(false);
pythonScriptInitialized.set(false);
- synchronized (statementFinishedNotifier) {
- statementFinishedNotifier.notify();
- }
}
@Override
@@ -609,9 +597,6 @@ public class PythonInterpreter extends Interpreter implements ExecuteResultHandl
LOGGER.error("python process failed", e);
pythonScriptRunning.set(false);
pythonScriptInitialized.set(false);
- synchronized (statementFinishedNotifier) {
- statementFinishedNotifier.notify();
- }
}
// Called by Python Process, used for debugging purpose
diff --git a/python/src/test/java/org/apache/zeppelin/python/BasePythonInterpreterTest.java b/python/src/test/java/org/apache/zeppelin/python/BasePythonInterpreterTest.java
index 6e8bbc9..a51c053 100644
--- a/python/src/test/java/org/apache/zeppelin/python/BasePythonInterpreterTest.java
+++ b/python/src/test/java/org/apache/zeppelin/python/BasePythonInterpreterTest.java
@@ -17,7 +17,6 @@
package org.apache.zeppelin.python;
-import net.jodah.concurrentunit.ConcurrentTestCase;
import org.apache.zeppelin.display.ui.CheckBox;
import org.apache.zeppelin.display.ui.Password;
import org.apache.zeppelin.display.ui.Select;
@@ -42,7 +41,7 @@ import static junit.framework.TestCase.assertTrue;
import static org.junit.Assert.assertEquals;
import static org.mockito.Mockito.mock;
-public abstract class BasePythonInterpreterTest extends ConcurrentTestCase {
+public abstract class BasePythonInterpreterTest {
protected InterpreterGroup intpGroup;
protected Interpreter interpreter;
diff --git a/python/src/test/java/org/apache/zeppelin/python/IPythonInterpreterTest.java b/python/src/test/java/org/apache/zeppelin/python/IPythonInterpreterTest.java
index 9eba8d8..28e6270 100644
--- a/python/src/test/java/org/apache/zeppelin/python/IPythonInterpreterTest.java
+++ b/python/src/test/java/org/apache/zeppelin/python/IPythonInterpreterTest.java
@@ -17,8 +17,6 @@
package org.apache.zeppelin.python;
-import net.jodah.concurrentunit.Waiter;
-import org.apache.commons.lang3.exception.ExceptionUtils;
import org.apache.zeppelin.interpreter.Interpreter;
import org.apache.zeppelin.interpreter.InterpreterContext;
import org.apache.zeppelin.interpreter.InterpreterException;
@@ -32,7 +30,6 @@ import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Properties;
-import java.util.concurrent.TimeoutException;
import static junit.framework.TestCase.assertTrue;
import static org.junit.Assert.assertEquals;
@@ -238,30 +235,4 @@ public class IPythonInterpreterTest extends BasePythonInterpreterTest {
assertEquals(InterpreterResult.Code.SUCCESS, result.code());
}
- @Test
- public void testIPythonProcessKilled() throws InterruptedException, TimeoutException {
- final Waiter waiter = new Waiter();
- Thread thread = new Thread() {
- @Override
- public void run() {
- try {
- InterpreterResult result = interpreter.interpret("import time\ntime.sleep(1000)",
- getInterpreterContext());
- waiter.assertEquals(InterpreterResult.Code.ERROR, result.code());
- waiter.assertEquals(
- "IPython kernel is abnormally exited, please check your code and log.",
- result.message().get(0).getData());
- } catch (InterpreterException e) {
- waiter.fail("Should not throw exception\n" + ExceptionUtils.getStackTrace(e));
- }
- waiter.resume();
- }
- };
- thread.start();
- Thread.sleep(3000);
- IPythonInterpreter iPythonInterpreter = (IPythonInterpreter)
- ((LazyOpenInterpreter) interpreter).getInnerInterpreter();
- iPythonInterpreter.getWatchDog().destroyProcess();
- waiter.await(3000);
- }
}
diff --git a/python/src/test/java/org/apache/zeppelin/python/PythonInterpreterTest.java b/python/src/test/java/org/apache/zeppelin/python/PythonInterpreterTest.java
index 19d2334..8748c00 100644
--- a/python/src/test/java/org/apache/zeppelin/python/PythonInterpreterTest.java
+++ b/python/src/test/java/org/apache/zeppelin/python/PythonInterpreterTest.java
@@ -17,8 +17,6 @@
package org.apache.zeppelin.python;
-import net.jodah.concurrentunit.Waiter;
-import org.apache.commons.lang3.exception.ExceptionUtils;
import org.apache.zeppelin.interpreter.Interpreter;
import org.apache.zeppelin.interpreter.InterpreterContext;
import org.apache.zeppelin.interpreter.InterpreterException;
@@ -30,7 +28,6 @@ import org.junit.Test;
import java.io.IOException;
import java.util.LinkedList;
import java.util.Properties;
-import java.util.concurrent.TimeoutException;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
@@ -41,7 +38,7 @@ import static org.junit.Assert.assertTrue;
public class PythonInterpreterTest extends BasePythonInterpreterTest {
-
+
@Override
public void setUp() throws InterpreterException {
@@ -53,7 +50,6 @@ public class PythonInterpreterTest extends BasePythonInterpreterTest {
properties.setProperty("zeppelin.python.gatewayserver_address", "127.0.0.1");
interpreter = new LazyOpenInterpreter(new PythonInterpreter(properties));
-
intpGroup.put("note", new LinkedList<Interpreter>());
intpGroup.get("note").add(interpreter);
interpreter.setInterpreterGroup(intpGroup);
@@ -109,31 +105,4 @@ public class PythonInterpreterTest extends BasePythonInterpreterTest {
t.join(2000);
assertFalse(t.isAlive());
}
-
- @Test
- public void testPythonProcessKilled() throws InterruptedException, TimeoutException {
- final Waiter waiter = new Waiter();
- Thread thread = new Thread() {
- @Override
- public void run() {
- try {
- InterpreterResult result = interpreter.interpret("import time\ntime.sleep(1000)",
- getInterpreterContext());
- waiter.assertEquals(InterpreterResult.Code.ERROR, result.code());
- waiter.assertEquals(
- "Python process is abnormally exited, please check your code and log.",
- result.message().get(0).getData());
- } catch (InterpreterException e) {
- waiter.fail("Should not throw exception\n" + ExceptionUtils.getStackTrace(e));
- }
- waiter.resume();
- }
- };
- thread.start();
- Thread.sleep(3000);
- PythonInterpreter pythonInterpreter = (PythonInterpreter)
- ((LazyOpenInterpreter) interpreter).getInnerInterpreter();
- pythonInterpreter.getPythonExecutor().getWatchdog().destroyProcess();
- waiter.await(3000);
- }
}
diff --git a/spark/interpreter/pom.xml b/spark/interpreter/pom.xml
index 9a35057..d7d8418 100644
--- a/spark/interpreter/pom.xml
+++ b/spark/interpreter/pom.xml
@@ -379,12 +379,6 @@
<scope>test</scope>
</dependency>
- <dependency>
- <groupId>net.jodah</groupId>
- <artifactId>concurrentunit</artifactId>
- <version>0.4.4</version>
- <scope>test</scope>
- </dependency>
</dependencies>
<build>
diff --git a/spark/interpreter/src/main/java/org/apache/zeppelin/spark/IPySparkInterpreter.java b/spark/interpreter/src/main/java/org/apache/zeppelin/spark/IPySparkInterpreter.java
index 594c171..7589895 100644
--- a/spark/interpreter/src/main/java/org/apache/zeppelin/spark/IPySparkInterpreter.java
+++ b/spark/interpreter/src/main/java/org/apache/zeppelin/spark/IPySparkInterpreter.java
@@ -83,8 +83,7 @@ public class IPySparkInterpreter extends IPythonInterpreter {
}
@Override
- public InterpreterResult interpret(String st,
- InterpreterContext context) throws InterpreterException {
+ public InterpreterResult interpret(String st, InterpreterContext context) {
InterpreterContext.set(context);
String jobGroupId = Utils.buildJobGroupId(context);
String jobDesc = Utils.buildJobDesc(context);
[zeppelin] 01/02: Revert "[ZEPPELIN-4089] handle ipython kernel
crash"
Posted by zj...@apache.org.
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
commit 6c24d1f31c289e015795bd9f74087d63a4a5ae39
Author: Jeff Zhang <zj...@apache.org>
AuthorDate: Fri Mar 29 22:19:50 2019 +0800
Revert "[ZEPPELIN-4089] handle ipython kernel crash"
This reverts commit ef5e173d3aa1e708cc8994a3f82a25d357ca7005.
---
.../main/resources/grpc/python/ipython_server.py | 44 ++++++----------------
.../zeppelin/python/IPythonInterpreterTest.java | 44 ----------------------
2 files changed, 12 insertions(+), 76 deletions(-)
diff --git a/python/src/main/resources/grpc/python/ipython_server.py b/python/src/main/resources/grpc/python/ipython_server.py
index 36e0a13..4b68efd 100644
--- a/python/src/main/resources/grpc/python/ipython_server.py
+++ b/python/src/main/resources/grpc/python/ipython_server.py
@@ -16,7 +16,6 @@
from __future__ import print_function
import jupyter_client
-import os
import sys
import threading
import time
@@ -26,6 +25,8 @@ import grpc
import ipython_pb2
import ipython_pb2_grpc
+_ONE_DAY_IN_SECONDS = 60 * 60 * 24
+
is_py2 = sys.version[0] == '2'
if is_py2:
import Queue as queue
@@ -33,6 +34,8 @@ else:
import queue as queue
+TIMEOUT = 60*60*24*365*100 # 100 years
+
class IPython(ipython_pb2_grpc.IPythonServicer):
def __init__(self, server):
@@ -70,16 +73,13 @@ class IPython(ipython_pb2_grpc.IPythonServicer):
def execute_worker():
reply = self._kc.execute_interactive(request.code,
output_hook=_output_hook,
- timeout=None)
+ timeout=TIMEOUT)
payload_reply.append(reply)
t = threading.Thread(name="ConsumerThread", target=execute_worker)
t.start()
- # We want to ensure that the kernel is alive because in case of OOM or other errors
- # Execution might be stuck there:
- # https://github.com/jupyter/jupyter_client/blob/master/jupyter_client/blocking/client.py#L32
- while t.is_alive() and self.isKernelAlive():
+ while t.is_alive():
while not stdout_queue.empty():
output = stdout_queue.get()
yield ipython_pb2.ExecuteResponse(status=ipython_pb2.SUCCESS,
@@ -96,14 +96,6 @@ class IPython(ipython_pb2_grpc.IPythonServicer):
type=ipython_pb2.IMAGE,
output=output)
- # if kernel is not alive (should be same as thread is still alive), means that we face
- # an unexpected issue.
- if not self.isKernelAlive() or t.is_alive():
- yield ipython_pb2.ExecuteResponse(status=ipython_pb2.ERROR,
- type=ipython_pb2.TEXT,
- output="Ipython kernel has been stopped. Please check logs. It might be because of an out of memory issue.")
- return
-
while not stdout_queue.empty():
output = stdout_queue.get()
yield ipython_pb2.ExecuteResponse(status=ipython_pb2.SUCCESS,
@@ -135,21 +127,15 @@ class IPython(ipython_pb2_grpc.IPythonServicer):
return ipython_pb2.CancelResponse()
def complete(self, request, context):
- reply = self._kc.complete(request.code, request.cursor, reply=True, timeout=None)
+ reply = self._kc.complete(request.code, request.cursor, reply=True, timeout=TIMEOUT)
return ipython_pb2.CompletionResponse(matches=reply['content']['matches'])
def status(self, request, context):
return ipython_pb2.StatusResponse(status = self._status)
- def isKernelAlive(self):
- return self._km.is_alive()
-
- def terminate(self):
- self._km.shutdown_kernel()
-
def stop(self, request, context):
- self.terminate()
- return ipython_pb2.StopResponse()
+ self._server.stop(0)
+ sys.exit(0)
def serve(port):
@@ -160,16 +146,10 @@ def serve(port):
server.start()
ipython.start()
try:
- while ipython.isKernelAlive():
- time.sleep(5)
+ while True:
+ time.sleep(_ONE_DAY_IN_SECONDS)
except KeyboardInterrupt:
- print("interrupted")
- finally:
- print("shutdown")
- # we let 2 sc for all request to be complete
- server.stop(2)
- ipython.terminate()
- os._exit(0)
+ server.stop(0)
if __name__ == '__main__':
serve(sys.argv[1])
diff --git a/python/src/test/java/org/apache/zeppelin/python/IPythonInterpreterTest.java b/python/src/test/java/org/apache/zeppelin/python/IPythonInterpreterTest.java
index ca54502..9eba8d8 100644
--- a/python/src/test/java/org/apache/zeppelin/python/IPythonInterpreterTest.java
+++ b/python/src/test/java/org/apache/zeppelin/python/IPythonInterpreterTest.java
@@ -24,7 +24,6 @@ import org.apache.zeppelin.interpreter.InterpreterContext;
import org.apache.zeppelin.interpreter.InterpreterException;
import org.apache.zeppelin.interpreter.InterpreterGroup;
import org.apache.zeppelin.interpreter.InterpreterResult;
-import org.apache.zeppelin.interpreter.InterpreterResult.Code;
import org.apache.zeppelin.interpreter.InterpreterResultMessage;
import org.apache.zeppelin.interpreter.LazyOpenInterpreter;
import org.junit.Test;
@@ -70,49 +69,6 @@ public class IPythonInterpreterTest extends BasePythonInterpreterTest {
}
@Test
- public void testIpythonKernelCrash_shouldNotHangExecution()
- throws InterpreterException, IOException {
- // The goal of this test is to ensure that we handle case when the kernel die.
- // In order to do so, we will kill the kernel process from the python code.
- // A real example of that could be a out of memory by the code we execute.
- String codeDep = "!pip install psutil";
- String codeFindPID = "from os import getpid\n"
- + "import psutil\n"
- + "pids = psutil.pids()\n"
- + "my_pid = getpid()\n"
- + "pidToKill = []\n"
- + "for pid in pids:\n"
- + " try:\n"
- + " p = psutil.Process(pid)\n"
- + " cmd = p.cmdline()\n"
- + " for arg in cmd:\n"
- + " if arg.count('ipykernel'):\n"
- + " pidToKill.append(pid)\n"
- + " except:\n"
- + " continue\n"
- + "len(pidToKill)";
- String codeKillKernel = "from os import kill\n"
- + "import signal\n"
- + "for pid in pidToKill:\n"
- + " kill(pid, signal.SIGKILL)";
- InterpreterContext context = getInterpreterContext();
- InterpreterResult result = interpreter.interpret(codeDep, context);
- assertEquals(InterpreterResult.Code.SUCCESS, result.code());
- context = getInterpreterContext();
- result = interpreter.interpret(codeFindPID, context);
- assertEquals(Code.SUCCESS, result.code());
- InterpreterResultMessage output = context.out.toInterpreterResultMessage().get(0);
- int numberOfPID = Integer.parseInt(output.getData());
- assertTrue(numberOfPID > 0);
- context = getInterpreterContext();
- result = interpreter.interpret(codeKillKernel, context);
- assertEquals(Code.ERROR, result.code());
- output = context.out.toInterpreterResultMessage().get(0);
- assertTrue(output.getData().equals("Ipython kernel has been stopped. Please check logs. "
- + "It might be because of an out of memory issue."));
- }
-
- @Test
public void testIPythonAdvancedFeatures()
throws InterpreterException, InterruptedException, IOException {
// ipython help