k8s runtime: force deletion to avoid hung function worker during conn…

…ector restart (#12504)
apache · Nov 6, 2021 · a3f6aba · a3f6aba
1 parent 36f151c
commit a3f6aba
Show file tree

Hide file tree

Showing 5 changed files with 24 additions and 4 deletions.
diff --git a/conf/functions_worker.yml b/conf/functions_worker.yml
@@ -216,6 +216,10 @@ functionRuntimeFactoryConfigs:
 #    extraFunctionDependenciesDir:
 #    # Additional memory padding added on top of the memory requested by the function per on a per instance basis
 #    percentMemoryPadding: 10
+#    # The duration in seconds before the StatefulSet deleted on function stop/restart.
+#    # Value must be non-negative integer. The value zero indicates delete immediately.
+#    # Default is 5 seconds.
+#    gracePeriodSeconds: 5
 
 ## A set of the minimum amount of resources functions must request.
 ## Support for this depends on function runtime.

diff --git a/...ntime/src/main/java/org/apache/pulsar/functions/runtime/kubernetes/KubernetesRuntime.java b/...ntime/src/main/java/org/apache/pulsar/functions/runtime/kubernetes/KubernetesRuntime.java
@@ -149,6 +149,7 @@ public class KubernetesRuntime implements Runtime {
     private int percentMemoryPadding;
     private double cpuOverCommitRatio;
     private double memoryOverCommitRatio;
+    private int gracePeriodSeconds;
     private final Optional<KubernetesFunctionAuthProvider> functionAuthDataCacheProvider;
     private final AuthenticationConfig authConfig;
     private Integer grpcPort;
@@ -186,6 +187,7 @@ public class KubernetesRuntime implements Runtime {
                       int percentMemoryPadding,
                       double cpuOverCommitRatio,
                       double memoryOverCommitRatio,
+                      int gracePeriodSeconds,
                       Optional<KubernetesFunctionAuthProvider> functionAuthDataCacheProvider,
                       boolean authenticationEnabled,
                       Integer grpcPort,
@@ -212,6 +214,7 @@ public class KubernetesRuntime implements Runtime {
         this.percentMemoryPadding = percentMemoryPadding;
         this.cpuOverCommitRatio = cpuOverCommitRatio;
         this.memoryOverCommitRatio = memoryOverCommitRatio;
+        this.gracePeriodSeconds = gracePeriodSeconds;
         this.authenticationEnabled = authenticationEnabled;
         this.manifestCustomizer = manifestCustomizer;
         this.functionInstanceClassPath = functionInstanceClassPath;
@@ -567,7 +570,7 @@ private void submitStatefulSet() throws Exception {
     public void deleteStatefulSet() throws InterruptedException {
         String statefulSetName = createJobName(instanceConfig.getFunctionDetails(), this.jobName);
         final V1DeleteOptions options = new V1DeleteOptions();
-        options.setGracePeriodSeconds(5L);
+        options.setGracePeriodSeconds((long)gracePeriodSeconds);
         options.setPropagationPolicy("Foreground");
 
         String fqfn = FunctionCommon.getFullyQualifiedName(instanceConfig.getFunctionDetails());
@@ -583,8 +586,8 @@ public void deleteStatefulSet() throws InterruptedException {
                         response = appsClient.deleteNamespacedStatefulSetCall(
                                 statefulSetName,
                                 jobNamespace, null, null,
-                                5, null, "Foreground",
-                                null, null)
+                                gracePeriodSeconds, null, "Foreground",
+                                options, null)
                                 .execute();
                     } catch (ApiException e) {
                         // if already deleted
@@ -735,7 +738,7 @@ public void deleteService() throws InterruptedException {
                                 serviceName,
                                 jobNamespace, null, null,
                                 0, null,
-                                "Foreground", null, null).execute();
+                                "Foreground", options, null).execute();
                     } catch (ApiException e) {
                         // if already deleted
                         if (e.getCode() == HTTP_NOT_FOUND) {

diff --git a/...rc/main/java/org/apache/pulsar/functions/runtime/kubernetes/KubernetesRuntimeFactory.java b/...rc/main/java/org/apache/pulsar/functions/runtime/kubernetes/KubernetesRuntimeFactory.java
@@ -103,6 +103,7 @@ public class KubernetesRuntimeFactory implements RuntimeFactory {
     private String narExtractionDirectory;
     private String functionInstanceClassPath;
     private String downloadDirectory;
+    private int gracePeriodSeconds;
 
     @ToString.Exclude
     @EqualsAndHashCode.Exclude
@@ -200,6 +201,7 @@ public void initialize(WorkerConfig workerConfig, AuthenticationConfig authentic
         this.percentMemoryPadding = factoryConfig.getPercentMemoryPadding();
         this.cpuOverCommitRatio = factoryConfig.getCpuOverCommitRatio();
         this.memoryOverCommitRatio = factoryConfig.getMemoryOverCommitRatio();
+        this.gracePeriodSeconds = factoryConfig.getGracePeriodSeconds();
         this.pulsarServiceUrl = StringUtils.isEmpty(factoryConfig.getPulsarServiceUrl())
                 ? workerConfig.getPulsarServiceUrl() : factoryConfig.getPulsarServiceUrl();
         this.pulsarAdminUrl = StringUtils.isEmpty(factoryConfig.getPulsarAdminUrl())
@@ -318,6 +320,7 @@ public KubernetesRuntime createContainer(InstanceConfig instanceConfig, String c
             percentMemoryPadding,
             cpuOverCommitRatio,
             memoryOverCommitRatio,
+            gracePeriodSeconds,
             authProvider,
             authenticationEnabled,
             grpcPort,

diff --git a/...n/java/org/apache/pulsar/functions/runtime/kubernetes/KubernetesRuntimeFactoryConfig.java b/...n/java/org/apache/pulsar/functions/runtime/kubernetes/KubernetesRuntimeFactoryConfig.java
@@ -161,5 +161,11 @@ public class KubernetesRuntimeFactoryConfig {
             doc = "The classpath where function instance files stored"
     )
     private String functionInstanceClassPath = "";
+    @FieldContext(
+            doc = "The duration in seconds before the StatefulSet deleted on function stop/restart. " +
+                    "Value must be non-negative integer. The value zero indicates delete immediately. " +
+                    "Default is 5 seconds."
+    )
+    protected int gracePeriodSeconds = 5;
 
 }
diff --git a/site2/docs/functions-runtime.md b/site2/docs/functions-runtime.md
@@ -136,6 +136,10 @@ functionRuntimeFactoryConfigs:
   extraFunctionDependenciesDir:
   # Additional memory padding added on top of the memory requested by the function per on a per instance basis
   percentMemoryPadding: 10
+  # The duration (in seconds) before the StatefulSet is deleted after a function stops or restarts.
+  # Value must be a non-negative integer. 0 indicates the StatefulSet is deleted immediately.
+  # Default is 5 seconds.
+  gracePeriodSeconds: 5
 ```
 
 If you run functions worker embedded in a broker on Kubernetes, you can use the default settings.