Consolidate error handling to Spawner (apache#203)

srkukarni · sijie · commit 631f22f72680 · 2018-03-03T16:44:13.000-08:00
diff --git a/pulsar-functions/conf/function_worker.yml b/pulsar-functions/conf/function_worker.yml
@@ -44,8 +44,8 @@ metricsConfig:
 # threadContainerFactory:
 #   threadGroupName: "Thread Function Container Group"
 processContainerFactory:
-  livenessCheckInterval: 10
-
+  logDirectory:
+  
 schedulerClassName: "org.apache.pulsar.functions.worker.scheduler.RoundRobinScheduler"
 functionAssignmentTopicName: "assignments"
 failureCheckFreqMs: 30000
diff --git a/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/container/FunctionContainer.java b/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/container/FunctionContainer.java
@@ -28,12 +28,16 @@
  */
 public interface FunctionContainer {
 
-    void start() throws Exception;
+    void start();
 
     void join() throws Exception;
 
     void stop();
 
+    boolean isAlive();
+
+    Exception getDeathException();
+
     CompletableFuture<InstanceCommunication.FunctionStatus> getFunctionStatus();
 
     CompletableFuture<InstanceCommunication.MetricsData> getAndResetMetrics();
diff --git a/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/container/ProcessFunctionContainer.java b/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/container/ProcessFunctionContainer.java
@@ -54,17 +54,13 @@ class ProcessFunctionContainer implements FunctionContainer {
     private Exception startupException;
     private ManagedChannel channel;
     private InstanceControlGrpc.InstanceControlFutureStub stub;
-    private Timer alivenessTimer;
-    private int alivenessCheckInterval;
-    private volatile boolean markedKilled = false;
 
     ProcessFunctionContainer(InstanceConfig instanceConfig,
                              int maxBufferedTuples,
                              String instanceFile,
                              String logDirectory,
                              String codeFile,
-                             String pulsarServiceUrl,
-                             int alivenessCheckInterval) {
+                             String pulsarServiceUrl) {
         List<String> args = new LinkedList<>();
         if (instanceConfig.getFunctionConfig().getRuntime() == Function.FunctionConfig.Runtime.JAVA) {
             args.add("java");
@@ -168,27 +164,19 @@ class ProcessFunctionContainer implements FunctionContainer {
         args.add(String.valueOf(instancePort));
 
         processBuilder = new ProcessBuilder(args);
-        this.alivenessCheckInterval = alivenessCheckInterval;
     }
 
     /**
      * The core logic that initialize the thread container and executes the function
      */
     @Override
-    public void start() throws Exception {
+    public void start() {
         Runtime.getRuntime().addShutdownHook(new Thread(() -> process.destroy()));
         startProcess();
         channel = ManagedChannelBuilder.forAddress("127.0.0.1", instancePort)
                 .usePlaintext(true)
                 .build();
         stub = InstanceControlGrpc.newFutureStub(channel);
-        alivenessTimer = new Timer();
-        alivenessTimer.scheduleAtFixedRate(new TimerTask() {
-            @Override
-            public void run() {
-                checkForAliveness();
-            }
-        }, alivenessCheckInterval * 1000, alivenessCheckInterval * 1000);
     }
 
     @Override
@@ -197,11 +185,9 @@ public void join() throws Exception {
     }
 
     @Override
-    public synchronized void stop() {
-        markedKilled = true;
+    public void stop() {
         process.destroy();
         channel.shutdown();
-        alivenessTimer.cancel();
     }
 
     @Override
@@ -262,6 +248,7 @@ private int findAvailablePort() {
     }
 
     private void startProcess() {
+        startupException = null;
         try {
             log.info("ProcessBuilder starting the process with args {}", String.join(" ", processBuilder.command()));
             process = processBuilder.start();
@@ -278,21 +265,24 @@ private void startProcess() {
         }
     }
 
-    private synchronized void checkForAliveness() {
-        if (markedKilled) return;
-        if (!process.isAlive()) {
-            log.error("Process is no longer alive, Restarting...");
-            InputStream errorStream = process.getErrorStream();
-            try {
-                byte[] errorBytes = new byte[errorStream.available()];
-                errorStream.read(errorBytes);
-                String errorMessage = new String(errorBytes);
-                startupException = new RuntimeException(errorMessage);
-                log.error("ErrorStream was " + errorMessage);
-            } catch (Exception ex) {
-                startupException = ex;
-            }
-            startProcess();
+    @Override
+    public boolean isAlive() {
+        return process.isAlive();
+    }
+
+    @Override
+    public Exception getDeathException() {
+        if (isAlive()) return null;
+        if (startupException != null) return startupException;
+        InputStream errorStream = process.getErrorStream();
+        try {
+            byte[] errorBytes = new byte[errorStream.available()];
+            errorStream.read(errorBytes);
+            String errorMessage = new String(errorBytes);
+            startupException = new RuntimeException(errorMessage);
+        } catch (Exception ex) {
+            startupException = ex;
         }
+        return startupException;
     }
 }
diff --git a/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/container/ProcessFunctionContainerFactory.java b/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/container/ProcessFunctionContainerFactory.java
@@ -30,22 +30,19 @@
 public class ProcessFunctionContainerFactory implements FunctionContainerFactory {
 
     private int maxBufferedTuples;
-    private int livenessCheckInterval;
     private String pulsarServiceUrl;
     private String javaInstanceJarFile;
     private String pythonInstanceFile;
     private String logDirectory;
 
     @VisibleForTesting
     public ProcessFunctionContainerFactory(int maxBufferedTuples,
-                                           int livenessCheckInterval,
                                            String pulsarServiceUrl,
                                            String javaInstanceJarFile,
                                            String pythonInstanceFile,
                                            String logDirectory) {
 
         this.maxBufferedTuples = maxBufferedTuples;
-        this.livenessCheckInterval = livenessCheckInterval;
         this.pulsarServiceUrl = pulsarServiceUrl;
         this.javaInstanceJarFile = javaInstanceJarFile;
         this.pythonInstanceFile = pythonInstanceFile;
@@ -71,8 +68,7 @@ public ProcessFunctionContainer createContainer(InstanceConfig instanceConfig, S
             instanceFile,
             logDirectory,
             codeFile,
-            pulsarServiceUrl,
-            livenessCheckInterval);
+            pulsarServiceUrl);
     }
 
     @Override
diff --git a/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/container/ThreadFunctionContainer.java b/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/container/ThreadFunctionContainer.java
@@ -44,6 +44,7 @@ class ThreadFunctionContainer implements FunctionContainer {
     @Getter
     private InstanceConfig instanceConfig;
     private JavaInstanceRunnable javaInstanceRunnable;
+    private Exception startupException;
 
     ThreadFunctionContainer(InstanceConfig instanceConfig,
                             int maxBufferedTuples,
@@ -73,6 +74,13 @@ class ThreadFunctionContainer implements FunctionContainer {
     @Override
     public void start() {
         log.info("ThreadContainer starting function with instance config {}", instanceConfig);
+        startupException = null;
+        this.fnThread.setUncaughtExceptionHandler(new Thread.UncaughtExceptionHandler() {
+            @Override
+            public void uncaughtException(Thread t, Throwable e) {
+                startupException = new Exception(e);
+            }
+        });
         this.fnThread.start();
     }
 
@@ -109,4 +117,15 @@ public CompletableFuture<FunctionStatus> getFunctionStatus() {
     public CompletableFuture<InstanceCommunication.MetricsData> getAndResetMetrics() {
         return CompletableFuture.completedFuture(javaInstanceRunnable.getAndResetMetrics());
     }
+
+    @Override
+    public boolean isAlive() {
+        return this.fnThread.isAlive();
+    }
+
+    @Override
+    public Exception getDeathException() {
+        if (isAlive()) return null;
+        else return startupException;
+    }
 }
diff --git a/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/spawner/Spawner.java b/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/spawner/Spawner.java
@@ -95,13 +95,19 @@ public void start() throws Exception {
             metricsCollectionTimer.scheduleAtFixedRate(new TimerTask() {
                 @Override
                 public void run() {
-                    log.info("Collecting metrics for function" + FunctionConfigUtils.getFullyQualifiedName(assignmentInfo.getFunctionConfig()));
-                    functionContainer.getAndResetMetrics().thenAccept(t -> {
-                        if (t != null) {
-                            log.debug("Collected metrics {}", t);
-                            metricsSink.processRecord(t, assignmentInfo.getFunctionConfig());
-                        }
-                    });
+                    if (functionContainer.isAlive()) {
+                        log.info("Collecting metrics for function" + FunctionConfigUtils.getFullyQualifiedName(assignmentInfo.getFunctionConfig()));
+                        functionContainer.getAndResetMetrics().thenAccept(t -> {
+                            if (t != null) {
+                                log.debug("Collected metrics {}", t);
+                                metricsSink.processRecord(t, assignmentInfo.getFunctionConfig());
+                            }
+                        });
+                    } else {
+                        log.error("Function Container is dead with exception", functionContainer.getDeathException());
+                        log.error("Restarting...");
+                        functionContainer.start();
+                    }
                 }
             }, metricsCollectionInterval * 1000, metricsCollectionInterval * 1000);
         }
diff --git a/pulsar-functions/runtime/src/test/java/org/apache/pulsar/functions/runtime/container/ProcessFunctionContainerTest.java b/pulsar-functions/runtime/src/test/java/org/apache/pulsar/functions/runtime/container/ProcessFunctionContainerTest.java
@@ -58,7 +58,7 @@ public ProcessFunctionContainerTest() {
         this.pulsarServiceUrl = "pulsar://localhost:6670";
         this.logDirectory = "Users/user/logs";
         this.factory = new ProcessFunctionContainerFactory(
-            1024, 10, pulsarServiceUrl, javaInstanceJarFile, pythonInstanceFile, logDirectory);
+            1024, pulsarServiceUrl, javaInstanceJarFile, pythonInstanceFile, logDirectory);
     }
 
     @AfterMethod
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
@@ -105,7 +105,6 @@ public FunctionRuntimeManager(WorkerConfig workerConfig,
         } else if (workerConfig.getProcessContainerFactory() != null) {
             this.functionContainerFactory = new ProcessFunctionContainerFactory(
                     workerConfig.getLimitsConfig().getMaxBufferedTuples(),
-                    workerConfig.getProcessContainerFactory().getLivenessCheckInterval(),
                     workerConfig.getPulsarServiceUrl(),
                     workerConfig.getProcessContainerFactory().getJavaInstanceJarLocation(),
                     workerConfig.getProcessContainerFactory().getPythonInstanceLocation(),
diff --git a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/WorkerConfig.java b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/WorkerConfig.java
@@ -82,7 +82,6 @@ static class ProcessContainerFactory {
         private String javaInstanceJarLocation;
         private String pythonInstanceLocation;
         private String logDirectory;
-        private int livenessCheckInterval;
     }
     private ProcessContainerFactory processContainerFactory;
 

Original file line number	Diff line number	Diff line change
`@@ -58,7 +58,7 @@ public ProcessFunctionContainerTest() {`
`58`	`58`	`this.pulsarServiceUrl = "pulsar://localhost:6670";`
`59`	`59`	`this.logDirectory = "Users/user/logs";`
`60`	`60`	`this.factory = new ProcessFunctionContainerFactory(`
`61`		`- 1024, 10, pulsarServiceUrl, javaInstanceJarFile, pythonInstanceFile, logDirectory);`
	`61`	`+ 1024, pulsarServiceUrl, javaInstanceJarFile, pythonInstanceFile, logDirectory);`
`62`	`62`	`}`
`63`	`63`
`64`	`64`	`@AfterMethod`
Original file line number	Diff line number	Diff line change
`@@ -82,7 +82,6 @@ static class ProcessContainerFactory {`
`82`	`82`	`private String javaInstanceJarLocation;`
`83`	`83`	`private String pythonInstanceLocation;`
`84`	`84`	`private String logDirectory;`
`85`		`- private int livenessCheckInterval;`
`86`	`85`	`}`
`87`	`86`	`private ProcessContainerFactory processContainerFactory;`
`88`	`87`