Close interpretergroup session if interpreter process crashed

Reamer · Oct 10, 2024 · fc840dd · fc840dd
1 parent bdf5b06
commit fc840dd
Showing 3 changed files with 19 additions and 0 deletions.
diff --git a/...r-integration/src/test/java/org/apache/zeppelin/integration/ZeppelinSparkClusterTest.java b/...r-integration/src/test/java/org/apache/zeppelin/integration/ZeppelinSparkClusterTest.java
@@ -1141,6 +1141,8 @@ public void testFailtoLaunchSpark() throws IOException {
 //            "Actual error message: " + p1.getReturn().message().get(0).getData());
 
           // run it again, and get the same error
+          note.run(p.getId(), true);
+          assertEquals(Status.FINISHED, p.getStatus());
           note.run(p1.getId(), true);
           assertEquals(Status.ERROR, p1.getStatus());
           // depends on JVM language

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
@@ -460,6 +460,14 @@ public ManagedInterpreterGroup getOrCreateInterpreterGroup(ExecutionContext exec
             groupId, executionContext);
         ManagedInterpreterGroup intpGroup = createInterpreterGroup(groupId);
         interpreterGroups.put(groupId, intpGroup);
+      } else {
+        // Check for a crashed interpreter process and restart interpreterGroup in this case
+        ManagedInterpreterGroup interpreterGroup = interpreterGroups.get(groupId);
+        if (interpreterGroup.isInterpreterProcessCrashed()) {
+          interpreterGroup.close();
+          interpreterGroups.remove(interpreterGroup.getId());
+          return getOrCreateInterpreterGroup(executionContext);
+        }
       }
       return interpreterGroups.get(groupId);
     } finally {

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
@@ -60,6 +60,15 @@ public InterpreterSetting getInterpreterSetting() {
     return interpreterSetting;
   }
 
+  public boolean isInterpreterProcessCrashed() {
+    if (remoteInterpreterProcess == null) {
+      return false;
+    }
+    synchronized (interpreterProcessCreationLock) {
+      return !remoteInterpreterProcess.isRunning();
+    }
+  }
+
   public RemoteInterpreterProcess getOrCreateInterpreterProcess(String userName,
                                                                 Properties properties)
       throws IOException {