[zeppelin] branch master updated: [ZEPPELIN-5604] Interpreter scheduler may not be shutdown properly

[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


The following commit(s) were added to refs/heads/master by this push:
     new f85d415  [ZEPPELIN-5604] Interpreter scheduler may not be shutdown properly
f85d415 is described below

commit f85d415225860afa8ff472189893dcd7370d3310
Author: Jeff Zhang <zj...@apache.org>
AuthorDate: Tue Dec 7 23:56:50 2021 +0800

    [ZEPPELIN-5604] Interpreter scheduler may not be shutdown properly
    
    ### What is this PR for?
    
    The issue will cause the thread used by the interpreter scheduler is not released, so overtime the thread pool may be exhausted. The root cause is that some uncaught exception may happen when closing interpreter, which causes the interpreter scheduler shutdown is not invoked. This PR put the scheduler shutdown in the finally block, so that it is always called.
    
    ### What type of PR is it?
    [Bug Fix]
    
    ### Todos
    * [ ] - Task
    
    ### What is the Jira issue?
    * https://issues.apache.org/jira/browse/ZEPPELIN-5604
    
    ### How should this be tested?
    * CI pass
    
    ### 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 #4307 from zjffdu/ZEPPELIN-5604 and squashes the following commits:
    
    02d9a61fec [Jeff Zhang] [ZEPPELIN-5604] Interpreter scheduler may not be shutdown properly
---
 .../interpreter/ManagedInterpreterGroup.java       | 29 +++++++++++-----------
 1 file changed, 14 insertions(+), 15 deletions(-)

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 fb17542..51aeab7 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
@@ -144,29 +144,28 @@ public class ManagedInterpreterGroup extends InterpreterGroup {
 
   private void closeInterpreter(Interpreter interpreter) {
     Scheduler scheduler = interpreter.getScheduler();
-
-    if (Boolean.parseBoolean(
-            interpreter.getProperty("zeppelin.interpreter.close.cancel_job", "true"))) {
-      for (final Job job : scheduler.getAllJobs()) {
-        if (!job.isTerminated()) {
-          job.abort();
-          job.setStatus(Job.Status.ABORT);
-          LOGGER.info("Job {} aborted ", job.getJobName());
+    try {
+      if (Boolean.parseBoolean(
+              interpreter.getProperty("zeppelin.interpreter.close.cancel_job", "true"))) {
+        for (final Job job : scheduler.getAllJobs()) {
+          if (!job.isTerminated()) {
+            job.abort();
+            job.setStatus(Job.Status.ABORT);
+            LOGGER.info("Job {} aborted ", job.getJobName());
+          }
         }
+      } else {
+        LOGGER.info("Keep job running while closing interpreter: {}", interpreter.getClassName());
       }
-    } else {
-      LOGGER.info("Keep job running while closing interpreter: {}", interpreter.getClassName());
-    }
 
-    try {
       LOGGER.info("Trying to close interpreter {}", interpreter.getClassName());
       interpreter.close();
     } catch (InterpreterException e) {
       LOGGER.warn("Fail to close interpreter {}", interpreter.getClassName(), e);
+    } finally {
+      //TODO(zjffdu) move the close of schedule to Interpreter
+      SchedulerFactory.singleton().removeScheduler(scheduler.getName());
     }
-
-    //TODO(zjffdu) move the close of schedule to Interpreter
-    SchedulerFactory.singleton().removeScheduler(scheduler.getName());
   }
 
   public synchronized List<Interpreter> getOrCreateSession(String user, String sessionId) {