[incubator-pulsar] branch master updated: Fix: avoid closing RuntimeFactory when any function stops (#1988)

[sijie]

This is an automated email from the ASF dual-hosted git repository.

sijie pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-pulsar.git


The following commit(s) were added to refs/heads/master by this push:
     new bd00214  Fix: avoid closing RuntimeFactory when any function stops 
(#1988)
bd00214 is described below

commit bd00214cd664ab3adc0369605430fca3c236572e
Author: Rajan Dhabalia <rdhaba...@apache.org>
AuthorDate: Wed Jun 20 00:15:26 2018 -0700

    Fix: avoid closing RuntimeFactory when any function stops (#1988)
    
    * Fix: avoid closing RuntimeFactory when any function stops
    
    * close runtimeFactory for process-runtime
---
 .../java/org/apache/pulsar/functions/runtime/JavaInstanceMain.java  | 6 +++++-
 .../java/org/apache/pulsar/functions/runtime/RuntimeSpawner.java    | 5 -----
 .../org/apache/pulsar/functions/worker/FunctionRuntimeManager.java  | 3 +++
 3 files changed, 8 insertions(+), 6 deletions(-)

diff --git 
a/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/JavaInstanceMain.java
 
b/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/JavaInstanceMain.java
index 554f9ac..5a63ec3 100644
--- 
a/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/JavaInstanceMain.java
+++ 
b/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/JavaInstanceMain.java
@@ -155,6 +155,7 @@ public class JavaInstanceMain implements AutoCloseable {
 
     private Server server;
     private RuntimeSpawner runtimeSpawner;
+    private ThreadRuntimeFactory containerFactory;
     private Long lastHealthCheckTs = null;
     private ScheduledExecutorService timer;
 
@@ -222,7 +223,7 @@ public class JavaInstanceMain implements AutoCloseable {
         instanceConfig.setFunctionDetails(functionDetails);
         instanceConfig.setPort(port);
 
-        ThreadRuntimeFactory containerFactory = new 
ThreadRuntimeFactory("LocalRunnerThreadGroup", pulsarServiceUrl,
+        containerFactory = new ThreadRuntimeFactory("LocalRunnerThreadGroup", 
pulsarServiceUrl,
                 stateStorageServiceUrl,
                 
AuthenticationConfig.builder().clientAuthenticationPlugin(clientAuthenticationPlugin)
                         
.clientAuthenticationParameters(clientAuthenticationParameters).useTls(isTrue(useTls))
@@ -302,6 +303,9 @@ public class JavaInstanceMain implements AutoCloseable {
             if (timer != null) {
                 timer.shutdown();
             }
+            if (containerFactory != null) {
+                containerFactory.close();
+            }
         } catch (Exception ex) {
             System.err.println(ex);
         }
diff --git 
a/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/RuntimeSpawner.java
 
b/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/RuntimeSpawner.java
index 8b60ded..236042e 100644
--- 
a/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/RuntimeSpawner.java
+++ 
b/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/RuntimeSpawner.java
@@ -27,14 +27,12 @@ import java.io.IOException;
 import java.util.Timer;
 import java.util.TimerTask;
 import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.ExecutionException;
 
 import lombok.Getter;
 import lombok.extern.slf4j.Slf4j;
 
 import org.apache.commons.lang3.StringUtils;
 import org.apache.pulsar.functions.instance.InstanceConfig;
-import org.apache.pulsar.functions.proto.InstanceCommunication;
 import org.apache.pulsar.functions.proto.InstanceCommunication.FunctionStatus;
 import org.apache.pulsar.functions.utils.Utils;
 import static 
org.apache.pulsar.functions.proto.Function.FunctionDetails.Runtime.PYTHON;
@@ -130,9 +128,6 @@ public class RuntimeSpawner implements AutoCloseable {
             runtime.stop();
             runtime = null;
         }
-        if (runtimeFactory != null) {
-            runtimeFactory.close();
-        }
         if (processLivenessCheckTimer != null) {
             processLivenessCheckTimer.cancel();
             processLivenessCheckTimer = null;
diff --git 
a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/FunctionRuntimeManager.java
 
b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/FunctionRuntimeManager.java
index ba1a33d..2512295 100644
--- 
a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/FunctionRuntimeManager.java
+++ 
b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/FunctionRuntimeManager.java
@@ -521,6 +521,9 @@ public class FunctionRuntimeManager implements 
AutoCloseable{
     public void close() throws Exception {
         this.functionActioner.close();
         this.functionAssignmentTailer.close();
+        if (runtimeFactory != null) {
+            runtimeFactory.close();
+        }
     }
 
     private Map<String, Assignment> diff(Map<String, Assignment> 
assignmentMap1, Map<String, Assignment> assignmentMap2) {