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 c6c34f5 adding worker to instance health check (#1833)
c6c34f5 is described below
commit c6c34f53fe4eb453a1fc28c7d26e90115f12ca6f
Author: Boyang Jerry Peng <[email protected]>
AuthorDate: Wed May 23 11:09:55 2018 -0700
adding worker to instance health check (#1833)
* adding worker to instance health check
* cleaning up logs
* fix bug
* refactoring
* fixing bug
* add check for python
---
.../org/apache/pulsar/admin/cli/CmdFunctions.java | 2 +-
.../src/main/python/InstanceCommunication_pb2.py | 57 +++++++++++++++++---
.../main/python/InstanceCommunication_pb2_grpc.py | 20 +++++--
.../instance/src/main/python/python_instance.py | 22 ++++++++
.../instance/src/main/python/server.py | 3 ++
.../src/main/proto/InstanceCommunication.proto | 5 ++
.../pulsar/functions/runtime/JavaInstanceMain.java | 63 ++++++++++++++++++++--
.../pulsar/functions/runtime/ProcessRuntime.java | 57 +++++++++++++++++---
.../apache/pulsar/functions/runtime/Runtime.java | 2 +-
.../pulsar/functions/runtime/RuntimeSpawner.java | 5 ++
10 files changed, 214 insertions(+), 22 deletions(-)
diff --git
a/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdFunctions.java
b/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdFunctions.java
index 30103c7..eedbbea 100644
---
a/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdFunctions.java
+++
b/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdFunctions.java
@@ -1126,7 +1126,7 @@ public class CmdFunctions extends CmdBase {
instanceConfig,
userCodeFile,
containerFactory,
- 0);
+ 30000);
spawners.add(runtimeSpawner);
runtimeSpawner.start();
}
diff --git
a/pulsar-functions/instance/src/main/python/InstanceCommunication_pb2.py
b/pulsar-functions/instance/src/main/python/InstanceCommunication_pb2.py
index 0b1d45e..b4e814d 100644
--- a/pulsar-functions/instance/src/main/python/InstanceCommunication_pb2.py
+++ b/pulsar-functions/instance/src/main/python/InstanceCommunication_pb2.py
@@ -1,4 +1,3 @@
-#!/usr/bin/env python
#
# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements. See the NOTICE file
@@ -18,8 +17,6 @@
# under the License.
#
-# -*- encoding: utf-8 -*-
-
# Generated by the protocol buffer compiler. DO NOT EDIT!
# source: InstanceCommunication.proto
@@ -42,7 +39,7 @@ DESCRIPTOR = _descriptor.FileDescriptor(
name='InstanceCommunication.proto',
package='proto',
syntax='proto3',
-
serialized_pb=_b('\n\x1bInstanceCommunication.proto\x12\x05proto\x1a\x1bgoogle/protobuf/empty.proto\"\xa1\x05\n\x0e\x46unctionStatus\x12\x0f\n\x07running\x18\x01
\x01(\x08\x12\x18\n\x10\x66\x61ilureException\x18\x02
\x01(\t\x12\x13\n\x0bnumRestarts\x18\x03
\x01(\x03\x12\x14\n\x0cnumProcessed\x18\x04 \x01(\x03\x12
\n\x18numSuccessfullyProcessed\x18\x05
\x01(\x03\x12\x19\n\x11numUserExceptions\x18\x06
\x01(\x03\x12H\n\x14latestUserExceptions\x18\x07
\x03(\x0b\x32*.proto.FunctionStatus.Ex [...]
+
serialized_pb=_b('\n\x1bInstanceCommunication.proto\x12\x05proto\x1a\x1bgoogle/protobuf/empty.proto\"\xa1\x05\n\x0e\x46unctionStatus\x12\x0f\n\x07running\x18\x01
\x01(\x08\x12\x18\n\x10\x66\x61ilureException\x18\x02
\x01(\t\x12\x13\n\x0bnumRestarts\x18\x03
\x01(\x03\x12\x14\n\x0cnumProcessed\x18\x04 \x01(\x03\x12
\n\x18numSuccessfullyProcessed\x18\x05
\x01(\x03\x12\x19\n\x11numUserExceptions\x18\x06
\x01(\x03\x12H\n\x14latestUserExceptions\x18\x07
\x03(\x0b\x32*.proto.FunctionStatus.Ex [...]
,
dependencies=[google_dot_protobuf_dot_empty__pb2.DESCRIPTOR,])
@@ -394,6 +391,37 @@ _METRICSDATA = _descriptor.Descriptor(
serialized_end=1027,
)
+
+_HEALTHCHECKRESULT = _descriptor.Descriptor(
+ name='HealthCheckResult',
+ full_name='proto.HealthCheckResult',
+ filename=None,
+ file=DESCRIPTOR,
+ containing_type=None,
+ fields=[
+ _descriptor.FieldDescriptor(
+ name='success', full_name='proto.HealthCheckResult.success', index=0,
+ number=1, type=8, cpp_type=7, label=1,
+ has_default_value=False, default_value=False,
+ message_type=None, enum_type=None, containing_type=None,
+ is_extension=False, extension_scope=None,
+ options=None, file=DESCRIPTOR),
+ ],
+ extensions=[
+ ],
+ nested_types=[],
+ enum_types=[
+ ],
+ options=None,
+ is_extendable=False,
+ syntax='proto3',
+ extension_ranges=[],
+ oneofs=[
+ ],
+ serialized_start=1029,
+ serialized_end=1065,
+)
+
_FUNCTIONSTATUS_EXCEPTIONINFORMATION.containing_type = _FUNCTIONSTATUS
_FUNCTIONSTATUS_DESERIALIZATIONEXCEPTIONSENTRY.containing_type =
_FUNCTIONSTATUS
_FUNCTIONSTATUS.fields_by_name['latestUserExceptions'].message_type =
_FUNCTIONSTATUS_EXCEPTIONINFORMATION
@@ -407,6 +435,7 @@ _METRICSDATA.fields_by_name['metrics'].message_type =
_METRICSDATA_METRICSENTRY
DESCRIPTOR.message_types_by_name['FunctionStatus'] = _FUNCTIONSTATUS
DESCRIPTOR.message_types_by_name['FunctionStatusList'] = _FUNCTIONSTATUSLIST
DESCRIPTOR.message_types_by_name['MetricsData'] = _METRICSDATA
+DESCRIPTOR.message_types_by_name['HealthCheckResult'] = _HEALTHCHECKRESULT
_sym_db.RegisterFileDescriptor(DESCRIPTOR)
FunctionStatus = _reflection.GeneratedProtocolMessageType('FunctionStatus',
(_message.Message,), dict(
@@ -462,6 +491,13 @@ _sym_db.RegisterMessage(MetricsData)
_sym_db.RegisterMessage(MetricsData.DataDigest)
_sym_db.RegisterMessage(MetricsData.MetricsEntry)
+HealthCheckResult =
_reflection.GeneratedProtocolMessageType('HealthCheckResult',
(_message.Message,), dict(
+ DESCRIPTOR = _HEALTHCHECKRESULT,
+ __module__ = 'InstanceCommunication_pb2'
+ # @@protoc_insertion_point(class_scope:proto.HealthCheckResult)
+ ))
+_sym_db.RegisterMessage(HealthCheckResult)
+
DESCRIPTOR.has_options = True
DESCRIPTOR._options = _descriptor._ParseOptions(descriptor_pb2.FileOptions(),
_b('\n!org.apache.pulsar.functions.protoB\025InstanceCommunication'))
@@ -476,8 +512,8 @@ _INSTANCECONTROL = _descriptor.ServiceDescriptor(
file=DESCRIPTOR,
index=0,
options=None,
- serialized_start=1030,
- serialized_end=1185,
+ serialized_start=1068,
+ serialized_end=1290,
methods=[
_descriptor.MethodDescriptor(
name='GetFunctionStatus',
@@ -497,6 +533,15 @@ _INSTANCECONTROL = _descriptor.ServiceDescriptor(
output_type=_METRICSDATA,
options=None,
),
+ _descriptor.MethodDescriptor(
+ name='HealthCheck',
+ full_name='proto.InstanceControl.HealthCheck',
+ index=2,
+ containing_service=None,
+ input_type=google_dot_protobuf_dot_empty__pb2._EMPTY,
+ output_type=_HEALTHCHECKRESULT,
+ options=None,
+ ),
])
_sym_db.RegisterServiceDescriptor(_INSTANCECONTROL)
diff --git
a/pulsar-functions/instance/src/main/python/InstanceCommunication_pb2_grpc.py
b/pulsar-functions/instance/src/main/python/InstanceCommunication_pb2_grpc.py
index b37ee19..575f994 100644
---
a/pulsar-functions/instance/src/main/python/InstanceCommunication_pb2_grpc.py
+++
b/pulsar-functions/instance/src/main/python/InstanceCommunication_pb2_grpc.py
@@ -1,4 +1,3 @@
-#!/usr/bin/env python
#
# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements. See the NOTICE file
@@ -18,8 +17,6 @@
# under the License.
#
-# -*- encoding: utf-8 -*-
-
# Generated by the gRPC Python protocol compiler plugin. DO NOT EDIT!
import grpc
@@ -47,6 +44,11 @@ class InstanceControlStub(object):
request_serializer=google_dot_protobuf_dot_empty__pb2.Empty.SerializeToString,
response_deserializer=InstanceCommunication__pb2.MetricsData.FromString,
)
+ self.HealthCheck = channel.unary_unary(
+ '/proto.InstanceControl/HealthCheck',
+
request_serializer=google_dot_protobuf_dot_empty__pb2.Empty.SerializeToString,
+
response_deserializer=InstanceCommunication__pb2.HealthCheckResult.FromString,
+ )
class InstanceControlServicer(object):
@@ -67,6 +69,13 @@ class InstanceControlServicer(object):
context.set_details('Method not implemented!')
raise NotImplementedError('Method not implemented!')
+ def HealthCheck(self, request, context):
+ # missing associated documentation comment in .proto file
+ pass
+ context.set_code(grpc.StatusCode.UNIMPLEMENTED)
+ context.set_details('Method not implemented!')
+ raise NotImplementedError('Method not implemented!')
+
def add_InstanceControlServicer_to_server(servicer, server):
rpc_method_handlers = {
@@ -80,6 +89,11 @@ def add_InstanceControlServicer_to_server(servicer, server):
request_deserializer=google_dot_protobuf_dot_empty__pb2.Empty.FromString,
response_serializer=InstanceCommunication__pb2.MetricsData.SerializeToString,
),
+ 'HealthCheck': grpc.unary_unary_rpc_method_handler(
+ servicer.HealthCheck,
+
request_deserializer=google_dot_protobuf_dot_empty__pb2.Empty.FromString,
+
response_serializer=InstanceCommunication__pb2.HealthCheckResult.SerializeToString,
+ ),
}
generic_handler = grpc.method_handlers_generic_handler(
'proto.InstanceControl', rpc_method_handlers)
diff --git a/pulsar-functions/instance/src/main/python/python_instance.py
b/pulsar-functions/instance/src/main/python/python_instance.py
index 57bb292..6a82b0a 100644
--- a/pulsar-functions/instance/src/main/python/python_instance.py
+++ b/pulsar-functions/instance/src/main/python/python_instance.py
@@ -24,12 +24,15 @@
"""
import base64
import os
+import signal
import time
import Queue
import threading
from functools import partial
from collections import namedtuple
+from threading import Timer
import traceback
+import sys
import pulsar
import contextimpl
@@ -116,6 +119,21 @@ class PythonInstance(object):
self.contextimpl = None
self.total_stats = Stats()
self.current_stats = Stats()
+ self.last_health_check_ts = time.time()
+
+ def health_check(self):
+ self.last_health_check_ts = time.time()
+ health_check_result = InstanceCommunication_pb2.HealthCheckResult()
+ health_check_result.success = True
+ return health_check_result
+
+ def process_spawner_health_check_timer(self):
+ if time.time() - self.last_health_check_ts > 90:
+ Log.critical("Haven't received health check from spawner in a while.
Stopping instance...")
+ os.kill(os.getpid(), signal.SIGTERM)
+ sys.exit(1)
+
+ Timer(30, self.process_spawner_health_check_timer).start()
def run(self):
# Setup consumers and input deserializers
@@ -153,6 +171,10 @@ class PythonInstance(object):
self.exeuction_thread = threading.Thread(target=self.actual_execution)
self.exeuction_thread.start()
+ # start proccess spawner health check timer
+ self.last_health_check_ts = time.time()
+ Timer(30, self.process_spawner_health_check_timer).start()
+
def actual_execution(self):
Log.info("Started Thread for executing the function")
while True:
diff --git a/pulsar-functions/instance/src/main/python/server.py
b/pulsar-functions/instance/src/main/python/server.py
index 3fe7bd4..193e6ab 100644
--- a/pulsar-functions/instance/src/main/python/server.py
+++ b/pulsar-functions/instance/src/main/python/server.py
@@ -42,6 +42,9 @@ class
InstanceCommunicationServicer(InstanceCommunication_pb2_grpc.InstanceContr
Log.info("Came in GetAndResetMetrics")
return self.pyinstance.get_and_reset_metrics()
+ def HealthCheck(self, request, context):
+ return self.pyinstance.health_check()
+
def serve(port, pyinstance):
server_instance = grpc.server(futures.ThreadPoolExecutor(max_workers=10))
diff --git a/pulsar-functions/proto/src/main/proto/InstanceCommunication.proto
b/pulsar-functions/proto/src/main/proto/InstanceCommunication.proto
index 3ac0924..8db742b 100644
--- a/pulsar-functions/proto/src/main/proto/InstanceCommunication.proto
+++ b/pulsar-functions/proto/src/main/proto/InstanceCommunication.proto
@@ -65,7 +65,12 @@ message MetricsData {
map<string, DataDigest> metrics = 1;
}
+message HealthCheckResult {
+ bool success = 1;
+}
+
service InstanceControl {
rpc GetFunctionStatus(google.protobuf.Empty) returns (FunctionStatus) {}
rpc GetAndResetMetrics(google.protobuf.Empty) returns (MetricsData) {}
+ rpc HealthCheck(google.protobuf.Empty) returns (HealthCheckResult) {}
}
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 4e23038..c01f6bb 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
@@ -39,13 +39,17 @@ import
org.apache.pulsar.functions.proto.InstanceControlGrpc;
import java.lang.reflect.Type;
import java.util.Map;
+import java.util.TimerTask;
import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
/**
* A function container implemented using java thread.
*/
@Slf4j
-public class JavaInstanceMain {
+public class JavaInstanceMain implements AutoCloseable {
@Parameter(names = "--function_classname", description = "Function Class
Name\n", required = true)
protected String className;
@Parameter(
@@ -123,6 +127,9 @@ public class JavaInstanceMain {
protected String sinkSerdeClassName;
private Server server;
+ private RuntimeSpawner runtimeSpawner;
+ private Long lastHealthCheckTs = null;
+ private ScheduledExecutorService timer;
public JavaInstanceMain() { }
@@ -190,8 +197,7 @@ public class JavaInstanceMain {
"LocalRunnerThreadGroup",
pulsarServiceUrl,
stateStorageServiceUrl);
-
- RuntimeSpawner runtimeSpawner = new RuntimeSpawner(
+ runtimeSpawner = new RuntimeSpawner(
instanceConfig,
jarFile,
containerFactory,
@@ -216,9 +222,25 @@ public class JavaInstanceMain {
});
log.info("Starting runtimeSpawner");
runtimeSpawner.start();
+
+ timer = Executors.newSingleThreadScheduledExecutor();
+ timer.scheduleAtFixedRate(new TimerTask() {
+ @Override
+ public void run() {
+ try {
+ if (System.currentTimeMillis() - lastHealthCheckTs >
90000) {
+ 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);
+ }
+ }
+ }, 30000, 30000, TimeUnit.MILLISECONDS);
+
runtimeSpawner.join();
log.info("RuntimeSpawner quit, shutting down JavaInstance");
- server.shutdown();
+ close();
}
public static void main(String[] args) throws Exception {
@@ -231,11 +253,31 @@ public class JavaInstanceMain {
javaInstanceMain.start();
}
- static class InstanceControlImpl extends
InstanceControlGrpc.InstanceControlImplBase {
+ @Override
+ public void close() {
+ try {
+ // Use stderr here since the logger may have been reset by its JVM
shutdown hook.
+ if (server != null) {
+ server.shutdown();
+ }
+ if (runtimeSpawner != null) {
+ runtimeSpawner.close();
+ }
+ if (timer != null) {
+ timer.shutdown();
+ }
+ } catch (Exception ex) {
+ System.err.println(ex);
+ }
+ }
+
+
+ class InstanceControlImpl extends
InstanceControlGrpc.InstanceControlImplBase {
private RuntimeSpawner runtimeSpawner;
public InstanceControlImpl(RuntimeSpawner runtimeSpawner) {
this.runtimeSpawner = runtimeSpawner;
+ lastHealthCheckTs = System.currentTimeMillis();
}
@Override
@@ -266,5 +308,16 @@ public class JavaInstanceMain {
}
}
+ @Override
+ public void healthCheck(com.google.protobuf.Empty request,
+
io.grpc.stub.StreamObserver<org.apache.pulsar.functions.proto.InstanceCommunication.HealthCheckResult>
responseObserver) {
+ log.debug("Recieved health check request...");
+ InstanceCommunication.HealthCheckResult healthCheckResult
+ =
InstanceCommunication.HealthCheckResult.newBuilder().setSuccess(true).build();
+ responseObserver.onNext(healthCheckResult);
+ responseObserver.onCompleted();
+
+ lastHealthCheckTs = System.currentTimeMillis();
+ }
}
}
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 5b77eb2..977b9b1 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
@@ -26,7 +26,6 @@ import com.google.gson.Gson;
import com.google.protobuf.Empty;
import io.grpc.ManagedChannel;
import io.grpc.ManagedChannelBuilder;
-import java.util.concurrent.ExecutionException;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;
import org.apache.pulsar.functions.instance.InstanceConfig;
@@ -35,12 +34,12 @@ import
org.apache.pulsar.functions.proto.InstanceCommunication;
import org.apache.pulsar.functions.proto.InstanceCommunication.FunctionStatus;
import org.apache.pulsar.functions.proto.InstanceControlGrpc;
-import java.io.IOException;
import java.io.InputStream;
-import java.net.ServerSocket;
import java.util.*;
import java.util.concurrent.CompletableFuture;
-import
org.apache.pulsar.functions.proto.InstanceControlGrpc.InstanceControlFutureStub;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
/**
* A function container implemented using java thread.
@@ -58,12 +57,15 @@ class ProcessRuntime implements Runtime {
private Exception deathException;
private ManagedChannel channel;
private InstanceControlGrpc.InstanceControlFutureStub stub;
+ private ScheduledExecutorService timer;
+ private InstanceConfig instanceConfig;
ProcessRuntime(InstanceConfig instanceConfig,
String instanceFile,
String logDirectory,
String codeFile,
String pulsarServiceUrl) {
+ this.instanceConfig = instanceConfig;
this.instancePort = instanceConfig.getPort();
this.processArgs = composeArgs(instanceConfig, instanceFile,
logDirectory, codeFile, pulsarServiceUrl);
}
@@ -205,6 +207,21 @@ class ProcessRuntime implements Runtime {
.usePlaintext(true)
.build();
stub = InstanceControlGrpc.newFutureStub(channel);
+
+ timer = Executors.newSingleThreadScheduledExecutor();
+ timer.scheduleAtFixedRate(new TimerTask() {
+
+ @Override
+ public void run() {
+ CompletableFuture<InstanceCommunication.HealthCheckResult>
result = healthCheck();
+ try {
+ } catch (Exception e) {
+ log.error("Health check failed for {}-{}",
+ instanceConfig.getFunctionDetails().getName(),
+ instanceConfig.getInstanceId(), e);
+ }
+ }
+ }, 30000, 30000, TimeUnit.MILLISECONDS);
}
}
@@ -215,8 +232,15 @@ class ProcessRuntime implements Runtime {
@Override
public void stop() {
- process.destroy();
- channel.shutdown();
+ if (timer != null) {
+ timer.shutdown();
+ }
+ if (process != null) {
+ process.destroy();
+ }
+ if (channel != null) {
+ channel.shutdown();
+ }
channel = null;
stub = null;
}
@@ -272,6 +296,27 @@ class ProcessRuntime implements Runtime {
return retval;
}
+ public CompletableFuture<InstanceCommunication.HealthCheckResult>
healthCheck() {
+ CompletableFuture<InstanceCommunication.HealthCheckResult> retval =
new CompletableFuture<>();
+ if (stub == null) {
+ retval.completeExceptionally(new RuntimeException("Not alive"));
+ return retval;
+ }
+ ListenableFuture<InstanceCommunication.HealthCheckResult> response =
stub.healthCheck(Empty.newBuilder().build());
+ Futures.addCallback(response, new
FutureCallback<InstanceCommunication.HealthCheckResult>() {
+ @Override
+ public void onFailure(Throwable throwable) {
+ retval.completeExceptionally(throwable);
+ }
+
+ @Override
+ public void onSuccess(InstanceCommunication.HealthCheckResult t) {
+ retval.complete(t);
+ }
+ });
+ return retval;
+ }
+
private void startProcess() {
deathException = null;
try {
diff --git
a/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/Runtime.java
b/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/Runtime.java
index 29a8e81..18421f4 100644
---
a/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/Runtime.java
+++
b/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/Runtime.java
@@ -41,5 +41,5 @@ public interface Runtime {
CompletableFuture<InstanceCommunication.FunctionStatus>
getFunctionStatus();
CompletableFuture<InstanceCommunication.MetricsData> getAndResetMetrics();
-
+
}
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 0994adc..60b41de 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,10 +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.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;
@@ -118,6 +120,9 @@ public class RuntimeSpawner implements AutoCloseable {
runtime.stop();
runtime = null;
}
+ if (runtimeFactory != null) {
+ runtimeFactory.close();
+ }
if (processLivenessCheckTimer != null) {
processLivenessCheckTimer.cancel();
processLivenessCheckTimer = null;
--
To stop receiving notification emails like this one, please contact
[email protected].
