This is an automated email from the ASF dual-hosted git repository. jerrypeng pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/master by this push: new ded9086 Consolidate timer threads in functions (#3109) ded9086 is described below commit ded9086237ef6f8cf5d17676a2056bb7ff2daad3 Author: Boyang Jerry Peng <jerry.boyang.p...@gmail.com> AuthorDate: Mon Dec 3 17:47:13 2018 -0800 Consolidate timer threads in functions (#3109) * Use instanceCache schedulerExecutorService for timer * fixing process runtime * addressing comments --- .../functions/instance/FunctionStatsManager.java | 5 ++- .../pulsar/functions/instance/InstanceCache.java | 13 +++++-- .../functions/instance/JavaInstanceRunnable.java | 2 +- .../pulsar/functions/runtime/JavaInstanceMain.java | 28 +++++++-------- .../pulsar/functions/runtime/ProcessRuntime.java | 27 +++++++------- .../pulsar/functions/runtime/RuntimeSpawner.java | 41 +++++++++++----------- 6 files changed, 61 insertions(+), 55 deletions(-) diff --git a/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/FunctionStatsManager.java b/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/FunctionStatsManager.java index f9a3c77..9059f79 100644 --- a/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/FunctionStatsManager.java +++ b/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/FunctionStatsManager.java @@ -393,6 +393,9 @@ public class FunctionStatsManager implements AutoCloseable { @Override public void close() { - scheduledFuture.cancel(false); + if (scheduledFuture != null) { + scheduledFuture.cancel(false); + scheduledFuture = null; + } } } diff --git a/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/InstanceCache.java b/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/InstanceCache.java index 937c273..fe7e049 100644 --- a/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/InstanceCache.java +++ b/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/InstanceCache.java @@ -18,17 +18,24 @@ */ package org.apache.pulsar.functions.instance; +import io.netty.util.concurrent.DefaultThreadFactory; +import lombok.Getter; + import java.util.concurrent.Executors; import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.ThreadFactory; public class InstanceCache { private static InstanceCache instance; - public final ScheduledExecutorService executor; + @Getter + private final ScheduledExecutorService scheduledExecutorService; private InstanceCache() { - executor = Executors.newSingleThreadScheduledExecutor(); + ThreadFactory namedThreadFactory = + new DefaultThreadFactory("function-timer-thread"); + scheduledExecutorService = Executors.newSingleThreadScheduledExecutor(namedThreadFactory); } public static InstanceCache getInstanceCache() { @@ -43,7 +50,7 @@ public class InstanceCache { public static void shutdown() { synchronized (InstanceCache.class) { if (instance != null) { - instance.executor.shutdown(); + instance.scheduledExecutorService.shutdown(); } instance = null; } diff --git a/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/JavaInstanceRunnable.java b/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/JavaInstanceRunnable.java index d302e6a..e55fabc 100644 --- a/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/JavaInstanceRunnable.java +++ b/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/JavaInstanceRunnable.java @@ -216,7 +216,7 @@ public class JavaInstanceRunnable implements AutoCloseable, Runnable { if (this.collectorRegistry == null) { this.collectorRegistry = new CollectorRegistry(); } - this.stats = new FunctionStatsManager(this.collectorRegistry, this.metricsLabels, this.instanceCache.executor); + this.stats = new FunctionStatsManager(this.collectorRegistry, this.metricsLabels, this.instanceCache.getScheduledExecutorService()); ContextImpl contextImpl = setupContext(); javaInstance = setupJavaInstance(contextImpl); 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 f0be511..3a0a404 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 @@ -34,6 +34,7 @@ import io.prometheus.client.exporter.HTTPServer; import lombok.extern.slf4j.Slf4j; import org.apache.commons.lang3.StringUtils; import org.apache.pulsar.functions.instance.AuthenticationConfig; +import org.apache.pulsar.functions.instance.InstanceCache; import org.apache.pulsar.functions.instance.InstanceConfig; import org.apache.pulsar.functions.proto.Function.FunctionDetails; import org.apache.pulsar.functions.proto.InstanceCommunication; @@ -49,6 +50,7 @@ import java.util.TimerTask; import java.util.concurrent.ExecutionException; import java.util.concurrent.Executors; import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.ScheduledFuture; import java.util.concurrent.TimeUnit; /** @@ -122,8 +124,8 @@ public class JavaInstanceMain implements AutoCloseable { private RuntimeSpawner runtimeSpawner; private ThreadRuntimeFactory containerFactory; private Long lastHealthCheckTs = null; - private ScheduledExecutorService timer; private HTTPServer metricsServer; + private ScheduledFuture healthCheckTimer; public JavaInstanceMain() { } @@ -215,18 +217,14 @@ public class JavaInstanceMain implements AutoCloseable { metricsServer = new HTTPServer(new InetSocketAddress(metrics_port), collectorRegistry, true); if (expectedHealthCheckInterval > 0) { - timer = Executors.newSingleThreadScheduledExecutor(); - timer.scheduleAtFixedRate(new TimerTask() { - @Override - public void run() { - try { - if (System.currentTimeMillis() - lastHealthCheckTs > 3 * expectedHealthCheckInterval * 1000) { - log.info("Haven't received health check from spawner in a while. Stopping instance..."); - close(); - } - } catch (Exception e) { - log.error("Error occurred when checking for latest health check", e); + healthCheckTimer = InstanceCache.getInstanceCache().getScheduledExecutorService().scheduleAtFixedRate(() -> { + try { + if (System.currentTimeMillis() - lastHealthCheckTs > 3 * expectedHealthCheckInterval * 1000) { + log.info("Haven't received health check from spawner in a while. Stopping instance..."); + close(); } + } catch (Exception e) { + log.error("Error occurred when checking for latest health check", e); } }, expectedHealthCheckInterval * 1000, expectedHealthCheckInterval * 1000, TimeUnit.MILLISECONDS); } @@ -260,8 +258,8 @@ public class JavaInstanceMain implements AutoCloseable { if (runtimeSpawner != null) { runtimeSpawner.close(); } - if (timer != null) { - timer.shutdown(); + if (healthCheckTimer != null) { + healthCheckTimer.cancel(false); } if (containerFactory != null) { containerFactory.close(); @@ -269,6 +267,8 @@ public class JavaInstanceMain implements AutoCloseable { if (metricsServer != null) { metricsServer.stop(); } + + InstanceCache.shutdown(); } catch (Exception ex) { System.err.println(ex); } diff --git a/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/ProcessRuntime.java b/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/ProcessRuntime.java index 4edac17..70e7a3a 100644 --- a/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/ProcessRuntime.java +++ b/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/ProcessRuntime.java @@ -30,6 +30,7 @@ import lombok.Getter; import lombok.extern.slf4j.Slf4j; import org.apache.commons.lang3.StringUtils; import org.apache.pulsar.functions.instance.AuthenticationConfig; +import org.apache.pulsar.functions.instance.InstanceCache; import org.apache.pulsar.functions.instance.InstanceConfig; import org.apache.pulsar.functions.proto.Function.FunctionDetails; import org.apache.pulsar.functions.proto.InstanceCommunication; @@ -45,6 +46,7 @@ import java.util.TimerTask; import java.util.concurrent.CompletableFuture; import java.util.concurrent.Executors; import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.ScheduledFuture; import java.util.concurrent.TimeUnit; /** @@ -64,7 +66,7 @@ class ProcessRuntime implements Runtime { private Throwable deathException; private ManagedChannel channel; private InstanceControlGrpc.InstanceControlFutureStub stub; - private ScheduledExecutorService timer; + private ScheduledFuture timer; private InstanceConfig instanceConfig; private final Long expectedHealthCheckInterval; private final SecretsProviderConfigurator secretsProviderConfigurator; @@ -138,19 +140,14 @@ class ProcessRuntime implements Runtime { .build(); stub = InstanceControlGrpc.newFutureStub(channel); - timer = Executors.newSingleThreadScheduledExecutor(); - timer.scheduleAtFixedRate(new TimerTask() { - - @Override - public void run() { - CompletableFuture<InstanceCommunication.HealthCheckResult> result = healthCheck(); - try { - result.get(); - } catch (Exception e) { - log.error("Health check failed for {}-{}", - instanceConfig.getFunctionDetails().getName(), - instanceConfig.getInstanceId(), e); - } + timer = InstanceCache.getInstanceCache().getScheduledExecutorService().scheduleAtFixedRate(() -> { + CompletableFuture<InstanceCommunication.HealthCheckResult> result = healthCheck(); + try { + result.get(); + } catch (Exception e) { + log.error("Health check failed for {}-{}", + instanceConfig.getFunctionDetails().getName(), + instanceConfig.getInstanceId(), e); } }, expectedHealthCheckInterval, expectedHealthCheckInterval, TimeUnit.SECONDS); } @@ -164,7 +161,7 @@ class ProcessRuntime implements Runtime { @Override public void stop() { if (timer != null) { - timer.shutdown(); + timer.cancel(false); } if (process != null) { process.destroyForcibly(); 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 6b5abce..aa1784a 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,12 +27,15 @@ import java.io.IOException; import java.util.Timer; import java.util.TimerTask; import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ScheduledFuture; +import java.util.concurrent.TimeUnit; import lombok.Getter; import lombok.extern.slf4j.Slf4j; import org.apache.commons.lang3.StringUtils; import org.apache.pulsar.common.util.FutureUtil; +import org.apache.pulsar.functions.instance.InstanceCache; import org.apache.pulsar.functions.instance.InstanceConfig; import org.apache.pulsar.functions.proto.Function.FunctionDetails; import org.apache.pulsar.functions.proto.InstanceCommunication.FunctionStatus; @@ -50,7 +53,7 @@ public class RuntimeSpawner implements AutoCloseable { @Getter private Runtime runtime; - private Timer processLivenessCheckTimer; + private ScheduledFuture processLivenessCheckTimer; private int numRestarts; private long instanceLivenessCheckFreqMs; private Throwable runtimeDeathException; @@ -79,27 +82,23 @@ public class RuntimeSpawner implements AutoCloseable { // monitor function runtime to make sure it is running. If not, restart the function runtime if (!runtimeFactory.externallyManaged() && instanceLivenessCheckFreqMs > 0) { - processLivenessCheckTimer = new Timer(); - processLivenessCheckTimer.scheduleAtFixedRate(new TimerTask() { - @Override - public void run() { - Runtime runtime = RuntimeSpawner.this.runtime; - if (runtime != null && !runtime.isAlive()) { - log.error("{}/{}/{}-{} Function Container is dead with exception.. restarting", details.getTenant(), - details.getNamespace(), details.getName(), runtime.getDeathException()); - // Just for the sake of sanity, just destroy the runtime - try { - runtime.stop(); - runtimeDeathException = runtime.getDeathException(); - runtime.start(); - } catch (Exception e) { - log.error("{}/{}/{}-{} Function Restart failed", details.getTenant(), - details.getNamespace(), details.getName(), e, e); - } - numRestarts++; + processLivenessCheckTimer = InstanceCache.getInstanceCache().getScheduledExecutorService().scheduleAtFixedRate(() -> { + Runtime runtime = RuntimeSpawner.this.runtime; + if (runtime != null && !runtime.isAlive()) { + log.error("{}/{}/{}-{} Function Container is dead with exception.. restarting", details.getTenant(), + details.getNamespace(), details.getName(), runtime.getDeathException()); + // Just for the sake of sanity, just destroy the runtime + try { + runtime.stop(); + runtimeDeathException = runtime.getDeathException(); + runtime.start(); + } catch (Exception e) { + log.error("{}/{}/{}-{} Function Restart failed", details.getTenant(), + details.getNamespace(), details.getName(), e, e); } + numRestarts++; } - }, instanceLivenessCheckFreqMs, instanceLivenessCheckFreqMs); + }, instanceLivenessCheckFreqMs, instanceLivenessCheckFreqMs, TimeUnit.MILLISECONDS); } } @@ -139,7 +138,7 @@ public class RuntimeSpawner implements AutoCloseable { public void close() { // cancel liveness checker before stopping runtime. if (processLivenessCheckTimer != null) { - processLivenessCheckTimer.cancel(); + processLivenessCheckTimer.cancel(false); processLivenessCheckTimer = null; } if (null != runtime) {