You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by mm...@apache.org on 2019/02/15 23:46:02 UTC
[pulsar] branch master updated: fix: add retry and wait for
kubernetes creation/deletion (#3603)
This is an automated email from the ASF dual-hosted git repository.
mmerli 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 8a714ad fix: add retry and wait for kubernetes creation/deletion (#3603)
8a714ad is described below
commit 8a714ad26da8f51c4fe2f8fcb35f463585247934
Author: Boyang Jerry Peng <je...@gmail.com>
AuthorDate: Fri Feb 15 15:45:57 2019 -0800
fix: add retry and wait for kubernetes creation/deletion (#3603)
* fix: add retry and wait for kubernetes creation/deletion
* cleaning up
* adding retry logic for cleaning up subscriptions
* adjust number of retries
* correcting namespace
---
.../functions/runtime/KubernetesRuntime.java | 419 ++++++++++++++++-----
.../pulsar/functions/runtime/RuntimeUtils.java | 114 +++++-
.../pulsar/functions/runtime/RuntimeUtilsTest.java | 162 ++++++++
.../pulsar/functions/worker/FunctionActioner.java | 134 ++++---
4 files changed, 694 insertions(+), 135 deletions(-)
diff --git a/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/KubernetesRuntime.java b/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/KubernetesRuntime.java
index a0be580..705e2fa 100644
--- a/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/KubernetesRuntime.java
+++ b/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/KubernetesRuntime.java
@@ -28,6 +28,7 @@ import com.google.protobuf.Empty;
import com.squareup.okhttp.Response;
import io.grpc.ManagedChannel;
import io.grpc.ManagedChannelBuilder;
+import io.kubernetes.client.ApiException;
import io.kubernetes.client.apis.AppsV1Api;
import io.kubernetes.client.apis.CoreV1Api;
import io.kubernetes.client.custom.Quantity;
@@ -39,6 +40,7 @@ import io.kubernetes.client.models.V1EnvVarSource;
import io.kubernetes.client.models.V1LabelSelector;
import io.kubernetes.client.models.V1ObjectFieldSelector;
import io.kubernetes.client.models.V1ObjectMeta;
+import io.kubernetes.client.models.V1PodList;
import io.kubernetes.client.models.V1PodSpec;
import io.kubernetes.client.models.V1PodTemplateSpec;
import io.kubernetes.client.models.V1ResourceRequirements;
@@ -48,15 +50,6 @@ import io.kubernetes.client.models.V1ServiceSpec;
import io.kubernetes.client.models.V1StatefulSet;
import io.kubernetes.client.models.V1StatefulSetSpec;
import io.kubernetes.client.models.V1Toleration;
-
-import java.io.IOException;
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.LinkedList;
-import java.util.List;
-import java.util.Map;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;
import org.apache.pulsar.functions.instance.AuthenticationConfig;
@@ -67,14 +60,25 @@ import org.apache.pulsar.functions.proto.InstanceCommunication;
import org.apache.pulsar.functions.proto.InstanceCommunication.FunctionStatus;
import org.apache.pulsar.functions.proto.InstanceControlGrpc;
import org.apache.pulsar.functions.secretsproviderconfigurator.SecretsProviderConfigurator;
+import org.apache.pulsar.functions.utils.FunctionDetailsUtils;
import org.apache.pulsar.functions.utils.Utils;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import static java.net.HttpURLConnection.HTTP_CONFLICT;
+import static java.net.HttpURLConnection.HTTP_NOT_FOUND;
/**
* Kubernetes based runtime for running functions.
@@ -89,6 +93,9 @@ import static java.net.HttpURLConnection.HTTP_CONFLICT;
@VisibleForTesting
public class KubernetesRuntime implements Runtime {
+ private static int NUM_RETRIES = 5;
+ private static long SLEEP_BETWEEN_RETRIES_MS = 500;
+
private static final String ENV_SHARD_ID = "SHARD_ID";
private static final int maxJobNameSize = 55;
private static final Integer GRPC_PORT = 9093;
@@ -340,29 +347,44 @@ public class KubernetesRuntime implements Runtime {
final V1Service service = createService();
log.info("Submitting the following service to k8 {}", coreClient.getApiClient().getJSON().serialize(service));
- final Response response =
- coreClient.createNamespacedServiceCall(jobNamespace, service, null,
- null, null).execute();
- if (!response.isSuccessful()) {
- if (response.code() == HTTP_CONFLICT) {
- log.warn("Service already created for function {}/{}/{}",
- instanceConfig.getFunctionDetails().getTenant(),
- instanceConfig.getFunctionDetails().getNamespace(),
- instanceConfig.getFunctionDetails().getName());
- } else {
- log.error("Error creating Service for function {}/{}/{}:- {}",
- instanceConfig.getFunctionDetails().getTenant(),
- instanceConfig.getFunctionDetails().getNamespace(),
- instanceConfig.getFunctionDetails().getName(),
- response.message());
- // construct a message based on the k8s api server response
- throw new IllegalStateException(response.message());
- }
- } else {
- log.info("Service Created Successfully for function {}/{}/{}",
- instanceConfig.getFunctionDetails().getTenant(),
- instanceConfig.getFunctionDetails().getNamespace(),
- instanceConfig.getFunctionDetails().getName());
+ String fqfn = FunctionDetailsUtils.getFullyQualifiedName(instanceConfig.getFunctionDetails());
+
+ RuntimeUtils.Actions.Action createService = RuntimeUtils.Actions.Action.builder()
+ .actionName(String.format("Submitting service for function %s", fqfn))
+ .numRetries(NUM_RETRIES)
+ .sleepBetweenInvocationsMs(SLEEP_BETWEEN_RETRIES_MS)
+ .supplier(() -> {
+ final V1Service response;
+ try {
+ response = coreClient.createNamespacedService(jobNamespace, service, null);
+ } catch (ApiException e) {
+ // already exists
+ if (e.getCode() == HTTP_CONFLICT) {
+ log.warn("Service already present for function {}", fqfn);
+ return RuntimeUtils.Actions.ActionResult.builder().success(true).build();
+ }
+
+ String errorMsg = e.getResponseBody() != null ? e.getResponseBody() : e.getMessage();
+ return RuntimeUtils.Actions.ActionResult.builder()
+ .success(false)
+ .errorMsg(errorMsg)
+ .build();
+ }
+
+ return RuntimeUtils.Actions.ActionResult.builder().success(true).build();
+ })
+ .build();
+
+
+ AtomicBoolean success = new AtomicBoolean(false);
+ RuntimeUtils.Actions.newBuilder()
+ .addAction(createService.toBuilder()
+ .onSuccess(() -> success.set(true))
+ .build())
+ .run();
+
+ if (!success.get()) {
+ throw new RuntimeException(String.format("Failed to create service for function %s", fqfn));
}
}
@@ -397,75 +419,298 @@ public class KubernetesRuntime implements Runtime {
log.info("Submitting the following spec to k8 {}", appsClient.getApiClient().getJSON().serialize(statefulSet));
- final Response response =
- appsClient.createNamespacedStatefulSetCall(jobNamespace, statefulSet, null,
- null, null).execute();
- if (!response.isSuccessful()) {
- if (response.code() == HTTP_CONFLICT) {
- log.warn("Statefulset already present for function {}/{}/{}",
- instanceConfig.getFunctionDetails().getTenant(),
- instanceConfig.getFunctionDetails().getNamespace(),
- instanceConfig.getFunctionDetails().getName());
- } else {
- log.error("Error creating statefulset for function {}/{}/{}:- {}",
- instanceConfig.getFunctionDetails().getTenant(),
- instanceConfig.getFunctionDetails().getNamespace(),
- instanceConfig.getFunctionDetails().getName(),
- response.message());
- // construct a message based on the k8s api server response
- throw new IllegalStateException(response.message());
- }
- } else {
- log.info("Successfully created statefulset for function {}/{}/{}",
- instanceConfig.getFunctionDetails().getTenant(),
- instanceConfig.getFunctionDetails().getNamespace(),
- instanceConfig.getFunctionDetails().getName());
+ String fqfn = FunctionDetailsUtils.getFullyQualifiedName(instanceConfig.getFunctionDetails());
+
+ RuntimeUtils.Actions.Action createStatefulSet = RuntimeUtils.Actions.Action.builder()
+ .actionName(String.format("Submitting statefulset for function %s", fqfn))
+ .numRetries(NUM_RETRIES)
+ .sleepBetweenInvocationsMs(SLEEP_BETWEEN_RETRIES_MS)
+ .supplier(() -> {
+ final V1StatefulSet response;
+ try {
+ response = appsClient.createNamespacedStatefulSet(jobNamespace, statefulSet, null);
+ } catch (ApiException e) {
+ // already exists
+ if (e.getCode() == HTTP_CONFLICT) {
+ log.warn("Statefulset already present for function {}", fqfn);
+ return RuntimeUtils.Actions.ActionResult.builder().success(true).build();
+ }
+
+ String errorMsg = e.getResponseBody() != null ? e.getResponseBody() : e.getMessage();
+ return RuntimeUtils.Actions.ActionResult.builder()
+ .success(false)
+ .errorMsg(errorMsg)
+ .build();
+ }
+
+ return RuntimeUtils.Actions.ActionResult.builder().success(true).build();
+ })
+ .build();
+
+
+ AtomicBoolean success = new AtomicBoolean(false);
+ RuntimeUtils.Actions.newBuilder()
+ .addAction(createStatefulSet.toBuilder()
+ .onSuccess(() -> success.set(true))
+ .build())
+ .run();
+
+ if (!success.get()) {
+ throw new RuntimeException(String.format("Failed to create statefulset for function %s", fqfn));
}
}
- public void deleteStatefulSet() throws Exception {
+
+ public void deleteStatefulSet() throws InterruptedException {
+ String statefulSetName = createJobName(instanceConfig.getFunctionDetails());
final V1DeleteOptions options = new V1DeleteOptions();
options.setGracePeriodSeconds(0L);
options.setPropagationPolicy("Foreground");
- final Response response = appsClient.deleteNamespacedStatefulSetCall(
- createJobName(instanceConfig.getFunctionDetails()),
- jobNamespace, options, null, null, null, null, null, null)
- .execute();
- if (!response.isSuccessful()) {
- throw new RuntimeException(String.format("Error deleting statefulset for function {}/{}/{} :- {} ",
- instanceConfig.getFunctionDetails().getTenant(),
- instanceConfig.getFunctionDetails().getNamespace(),
- instanceConfig.getFunctionDetails().getName(),
- response.message()));
+ String fqfn = FunctionDetailsUtils.getFullyQualifiedName(instanceConfig.getFunctionDetails());
+ RuntimeUtils.Actions.Action deleteStatefulSet = RuntimeUtils.Actions.Action.builder()
+ .actionName(String.format("Deleting statefulset for function %s", fqfn))
+ .numRetries(NUM_RETRIES)
+ .sleepBetweenInvocationsMs(SLEEP_BETWEEN_RETRIES_MS)
+ .supplier(() -> {
+ Response response;
+ try {
+ // cannot use deleteNamespacedStatefulSet because of bug in kuberenetes
+ // https://github.com/kubernetes-client/java/issues/86
+ response = appsClient.deleteNamespacedStatefulSetCall(
+ statefulSetName,
+ jobNamespace, options, null,
+ null, null, null,
+ null, null)
+ .execute();
+ } catch (ApiException e) {
+ // if already deleted
+ if (e.getCode() == HTTP_NOT_FOUND) {
+ log.warn("Statefulset for function {} does not exist", fqfn);
+ return RuntimeUtils.Actions.ActionResult.builder().success(true).build();
+ }
+
+ String errorMsg = e.getResponseBody() != null ? e.getResponseBody() : e.getMessage();
+ return RuntimeUtils.Actions.ActionResult.builder()
+ .success(false)
+ .errorMsg(errorMsg)
+ .build();
+ } catch (IOException e) {
+ return RuntimeUtils.Actions.ActionResult.builder()
+ .success(false)
+ .errorMsg(e.getMessage())
+ .build();
+ }
+
+ // if already deleted
+ if (response.code() == HTTP_NOT_FOUND) {
+ log.warn("Statefulset for function {} does not exist", fqfn);
+ return RuntimeUtils.Actions.ActionResult.builder().success(true).build();
+ } else {
+ return RuntimeUtils.Actions.ActionResult.builder()
+ .success(response.isSuccessful())
+ .errorMsg(response.message())
+ .build();
+ }
+ })
+ .build();
+
+
+ RuntimeUtils.Actions.Action waitForStatefulSetDeletion = RuntimeUtils.Actions.Action.builder()
+ .actionName(String.format("Waiting for statefulset for function %s to complete deletion", fqfn))
+ .numRetries(NUM_RETRIES)
+ .sleepBetweenInvocationsMs(SLEEP_BETWEEN_RETRIES_MS)
+ .supplier(() -> {
+ V1StatefulSet response;
+ try {
+ response = appsClient.readNamespacedStatefulSet(jobNamespace, statefulSetName,
+ null, null, null);
+ } catch (ApiException e) {
+ // statefulset is gone
+ if (e.getCode() == HTTP_NOT_FOUND) {
+ return RuntimeUtils.Actions.ActionResult.builder().success(true).build();
+ }
+
+ String errorMsg = e.getResponseBody() != null ? e.getResponseBody() : e.getMessage();
+ return RuntimeUtils.Actions.ActionResult.builder()
+ .success(false)
+ .errorMsg(errorMsg)
+ .build();
+ }
+ return RuntimeUtils.Actions.ActionResult.builder()
+ .success(false)
+ .errorMsg(response.getStatus().toString())
+ .build();
+ })
+ .build();
+
+ // Need to wait for all pods to die so we can cleanup subscriptions.
+ RuntimeUtils.Actions.Action waitForStatefulPodsToTerminate = RuntimeUtils.Actions.Action.builder()
+ .actionName(String.format("Waiting for pods for function %s to terminate", fqfn))
+ .numRetries(NUM_RETRIES * 2)
+ .sleepBetweenInvocationsMs(SLEEP_BETWEEN_RETRIES_MS * 2)
+ .supplier(() -> {
+ String labels = String.format("tenant=%s,namespace=%s,name=%s",
+ instanceConfig.getFunctionDetails().getTenant(),
+ instanceConfig.getFunctionDetails().getNamespace(),
+ instanceConfig.getFunctionDetails().getName());
+
+ V1PodList response;
+ try {
+ response = coreClient.listNamespacedPod(jobNamespace, null, null,
+ null, null, labels,
+ null, null, null, null);
+ } catch (ApiException e) {
+
+ String errorMsg = e.getResponseBody() != null ? e.getResponseBody() : e.getMessage();
+ return RuntimeUtils.Actions.ActionResult.builder()
+ .success(false)
+ .errorMsg(errorMsg)
+ .build();
+ }
+
+ if (response.getItems().size() > 0) {
+ return RuntimeUtils.Actions.ActionResult.builder()
+ .success(false)
+ .errorMsg(response.getItems().size() + " pods still alive.")
+ .build();
+ } else {
+ return RuntimeUtils.Actions.ActionResult.builder()
+ .success(true)
+ .build();
+ }
+ })
+ .build();
+
+
+ AtomicBoolean success = new AtomicBoolean(false);
+ RuntimeUtils.Actions.newBuilder()
+ .addAction(deleteStatefulSet.toBuilder()
+ .continueOn(true)
+ .build())
+ .addAction(waitForStatefulSetDeletion.toBuilder()
+ .continueOn(false)
+ .onSuccess(() -> success.set(true))
+ .build())
+ .addAction(deleteStatefulSet.toBuilder()
+ .continueOn(true)
+ .build())
+ .addAction(waitForStatefulSetDeletion.toBuilder()
+ .onSuccess(() -> success.set(true))
+ .build())
+ .run();
+
+ if (!success.get()) {
+ throw new RuntimeException(String.format("Failed to delete statefulset for function %s", fqfn));
} else {
- log.info("Successfully deleted statefulset for function {}/{}/{}",
- instanceConfig.getFunctionDetails().getTenant(),
- instanceConfig.getFunctionDetails().getNamespace(),
- instanceConfig.getFunctionDetails().getName());
+ // wait for pods to terminate
+ RuntimeUtils.Actions.newBuilder()
+ .addAction(waitForStatefulPodsToTerminate)
+ .run();
}
}
- public void deleteService() throws Exception {
+ public void deleteService() throws InterruptedException {
+
final V1DeleteOptions options = new V1DeleteOptions();
options.setGracePeriodSeconds(0L);
options.setPropagationPolicy("Foreground");
- final Response response = coreClient.deleteNamespacedServiceCall(
- createJobName(instanceConfig.getFunctionDetails()),
- jobNamespace, options, null, null, null, null, null, null)
- .execute();
-
- if (!response.isSuccessful()) {
- throw new RuntimeException(String.format("Error deleting service for function {}/{}/{} :- {}",
- instanceConfig.getFunctionDetails().getTenant(),
- instanceConfig.getFunctionDetails().getNamespace(),
- instanceConfig.getFunctionDetails().getName(),
- response.message()));
- } else {
- log.info("Service deleted successfully for function {}/{}/{}",
- instanceConfig.getFunctionDetails().getTenant(),
- instanceConfig.getFunctionDetails().getNamespace(),
- instanceConfig.getFunctionDetails().getName());
+ String fqfn = FunctionDetailsUtils.getFullyQualifiedName(instanceConfig.getFunctionDetails());
+ String serviceName = createJobName(instanceConfig.getFunctionDetails());
+
+ RuntimeUtils.Actions.Action deleteService = RuntimeUtils.Actions.Action.builder()
+ .actionName(String.format("Deleting service for function %s", fqfn))
+ .numRetries(NUM_RETRIES)
+ .sleepBetweenInvocationsMs(SLEEP_BETWEEN_RETRIES_MS)
+ .supplier(() -> {
+ final Response response;
+ try {
+ // cannot use deleteNamespacedService because of bug in kuberenetes
+ // https://github.com/kubernetes-client/java/issues/86
+ response = coreClient.deleteNamespacedServiceCall(
+ serviceName,
+ jobNamespace, options, null,
+ null, null,
+ null, null, null).execute();
+ } catch (ApiException e) {
+ // if already deleted
+ if (e.getCode() == HTTP_NOT_FOUND) {
+ log.warn("Service for function {} does not exist", fqfn);
+ return RuntimeUtils.Actions.ActionResult.builder().success(true).build();
+ }
+
+ String errorMsg = e.getResponseBody() != null ? e.getResponseBody() : e.getMessage();
+ return RuntimeUtils.Actions.ActionResult.builder()
+ .success(false)
+ .errorMsg(errorMsg)
+ .build();
+ } catch (IOException e) {
+ return RuntimeUtils.Actions.ActionResult.builder()
+ .success(false)
+ .errorMsg(e.getMessage())
+ .build();
+ }
+
+ // if already deleted
+ if (response.code() == HTTP_NOT_FOUND) {
+ log.warn("Service for function {} does not exist", fqfn);
+ return RuntimeUtils.Actions.ActionResult.builder().success(true).build();
+ } else {
+ return RuntimeUtils.Actions.ActionResult.builder()
+ .success(response.isSuccessful())
+ .errorMsg(response.message())
+ .build();
+ }
+ })
+ .build();
+
+ RuntimeUtils.Actions.Action waitForServiceDeletion = RuntimeUtils.Actions.Action.builder()
+ .actionName(String.format("Waiting for statefulset for function %s to complete deletion", fqfn))
+ .numRetries(NUM_RETRIES)
+ .sleepBetweenInvocationsMs(SLEEP_BETWEEN_RETRIES_MS)
+ .supplier(() -> {
+ V1Service response;
+ try {
+ response = coreClient.readNamespacedService(serviceName, jobNamespace,
+ null, null, null);
+
+ } catch (ApiException e) {
+ // statefulset is gone
+ if (e.getCode() == HTTP_NOT_FOUND) {
+ return RuntimeUtils.Actions.ActionResult.builder().success(true).build();
+ }
+ String errorMsg = e.getResponseBody() != null ? e.getResponseBody() : e.getMessage();
+ return RuntimeUtils.Actions.ActionResult.builder()
+ .success(false)
+ .errorMsg(errorMsg)
+ .build();
+ }
+ return RuntimeUtils.Actions.ActionResult.builder()
+ .success(false)
+ .errorMsg(response.getStatus().toString())
+ .build();
+ })
+ .build();
+
+ AtomicBoolean success = new AtomicBoolean(false);
+ RuntimeUtils.Actions.newBuilder()
+ .addAction(deleteService.toBuilder()
+ .continueOn(true)
+ .build())
+ .addAction(waitForServiceDeletion.toBuilder()
+ .continueOn(false)
+ .onSuccess(() -> success.set(true))
+ .build())
+ .addAction(deleteService.toBuilder()
+ .continueOn(true)
+ .build())
+ .addAction(waitForServiceDeletion.toBuilder()
+ .onSuccess(() -> success.set(true))
+ .build())
+ .run();
+
+ if (!success.get()) {
+ throw new RuntimeException(String.format("Failed to delete service for function %s", fqfn));
}
}
diff --git a/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/RuntimeUtils.java b/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/RuntimeUtils.java
index e74ec20..77e1f8f 100644
--- a/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/RuntimeUtils.java
+++ b/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/RuntimeUtils.java
@@ -27,8 +27,13 @@ import java.io.InputStreamReader;
import java.net.HttpURLConnection;
import java.net.InetAddress;
import java.net.URL;
+import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
+import java.util.function.Supplier;
+
+import lombok.Builder;
+import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.pulsar.functions.instance.AuthenticationConfig;
@@ -37,6 +42,7 @@ import org.apache.pulsar.functions.proto.Function;
import org.apache.pulsar.functions.utils.FunctionDetailsUtils;
import org.apache.pulsar.functions.utils.functioncache.FunctionCacheEntry;
+import static org.apache.commons.lang3.StringUtils.isBlank;
import static org.apache.commons.lang3.StringUtils.isEmpty;
import static org.apache.commons.lang3.StringUtils.isNotBlank;
@@ -44,7 +50,7 @@ import static org.apache.commons.lang3.StringUtils.isNotBlank;
* Util class for common runtime functionality
*/
@Slf4j
-class RuntimeUtils {
+public class RuntimeUtils {
private static final String FUNCTIONS_EXTRA_DEPS_PROPERTY = "pulsar.functions.extra.dependencies.dir";
@@ -210,4 +216,110 @@ class RuntimeUtils {
rd.close();
return result.toString();
}
+
+ public static class Actions {
+ private List<Action> actions = new LinkedList<>();
+
+ @Data
+ @Builder(toBuilder=true)
+ public static class Action {
+ private String actionName;
+ private int numRetries = 1;
+ private Supplier<ActionResult> supplier;
+ private long sleepBetweenInvocationsMs = 500;
+ private Boolean continueOn;
+ private Runnable onFail;
+ private Runnable onSuccess;
+
+ public void verifyAction() {
+ if (isBlank(actionName)) {
+ throw new RuntimeException("Action name is empty!");
+ }
+ if (supplier == null) {
+ throw new RuntimeException("Supplier is not specified!");
+ }
+ }
+ }
+
+ @Data
+ @Builder
+ public static class ActionResult {
+ private boolean success;
+ private String errorMsg;
+ }
+
+ private Actions() {
+
+ }
+
+
+ public Actions addAction(Action action) {
+ action.verifyAction();
+ this.actions.add(action);
+ return this;
+ }
+
+ public static Actions newBuilder() {
+ return new Actions();
+ }
+
+ public int numActions() {
+ return actions.size();
+ }
+
+ public void run() throws InterruptedException {
+ Iterator<Action> it = this.actions.iterator();
+ while(it.hasNext()) {
+ Action action = it.next();
+
+ boolean success;
+ try {
+ success = runAction(action);
+ } catch (Exception e) {
+ log.error("Uncaught exception thrown when running action [ {} ]:", action.getActionName(), e);
+ success = false;
+ }
+ if (action.getContinueOn() != null
+ && success == action.getContinueOn()) {
+ continue;
+ } else {
+ // terminate
+ break;
+ }
+ }
+ }
+
+ private boolean runAction(Action action) throws InterruptedException {
+ for (int i = 0; i< action.getNumRetries(); i++) {
+
+ ActionResult actionResult = action.getSupplier().get();
+
+ if (actionResult.isSuccess()) {
+ log.info("Sucessfully completed action [ {} ]", action.getActionName());
+ if (action.getOnSuccess() != null) {
+ action.getOnSuccess().run();
+ }
+ return true;
+ } else {
+ if (actionResult.getErrorMsg() != null) {
+ log.warn("Error completing action [ {} ] :- {} - [ATTEMPT] {}/{}",
+ action.getActionName(),
+ actionResult.getErrorMsg(),
+ i + 1, action.getNumRetries());
+ } else {
+ log.warn("Error completing action [ {} ] [ATTEMPT] {}/{}",
+ action.getActionName(),
+ i + 1, action.getNumRetries());
+ }
+
+ Thread.sleep(action.sleepBetweenInvocationsMs);
+ }
+ }
+ log.error("Failed completing action [ {} ]. Giving up!", action.getActionName());
+ if (action.getOnFail() != null) {
+ action.getOnFail().run();
+ }
+ return false;
+ }
+ }
}
diff --git a/pulsar-functions/runtime/src/test/java/org/apache/pulsar/functions/runtime/RuntimeUtilsTest.java b/pulsar-functions/runtime/src/test/java/org/apache/pulsar/functions/runtime/RuntimeUtilsTest.java
new file mode 100644
index 0000000..431afcf
--- /dev/null
+++ b/pulsar-functions/runtime/src/test/java/org/apache/pulsar/functions/runtime/RuntimeUtilsTest.java
@@ -0,0 +1,162 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.functions.runtime;
+
+import org.junit.Assert;
+import org.testng.annotations.Test;
+
+import java.util.function.Supplier;
+
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+public class RuntimeUtilsTest {
+
+ @Test
+ public void testActions() throws InterruptedException {
+
+ // Test for success
+ Supplier<RuntimeUtils.Actions.ActionResult> supplier1 = mock(Supplier.class);
+ when(supplier1.get()).thenReturn(RuntimeUtils.Actions.ActionResult.builder().success(true).build());
+
+ Supplier<RuntimeUtils.Actions.ActionResult> supplier2 = mock(Supplier.class);
+ when(supplier2.get()).thenReturn(RuntimeUtils.Actions.ActionResult.builder().success(true).build());
+
+ Runnable onFail = mock(Runnable.class);
+ Runnable onSucess = mock(Runnable.class);
+
+ RuntimeUtils.Actions.Action action1 = spy(
+ RuntimeUtils.Actions.Action.builder()
+ .actionName("action1")
+ .numRetries(10)
+ .sleepBetweenInvocationsMs(100)
+ .supplier(supplier1)
+ .continueOn(true)
+ .onFail(onFail)
+ .onSuccess(onSucess)
+ .build());
+
+ RuntimeUtils.Actions.Action action2 = spy(
+ RuntimeUtils.Actions.Action.builder()
+ .actionName("action2")
+ .numRetries(20)
+ .sleepBetweenInvocationsMs(200)
+ .supplier(supplier2)
+ .build());
+
+ RuntimeUtils.Actions actions = RuntimeUtils.Actions.newBuilder()
+ .addAction(action1)
+ .addAction(action2);
+ actions.run();
+
+ Assert.assertEquals(actions.numActions(), 2);
+ verify(supplier1, times(1)).get();
+ verify(onFail, times(0)).run();
+ verify(onSucess, times(1)).run();
+ verify(supplier2, times(1)).get();
+
+ // test only run 1 action
+
+ supplier1 = mock(Supplier.class);
+ when(supplier1.get()).thenReturn(RuntimeUtils.Actions.ActionResult.builder().success(true).build());
+
+ supplier2 = mock(Supplier.class);
+ when(supplier2.get()).thenReturn(RuntimeUtils.Actions.ActionResult.builder().success(true).build());
+
+ onFail = mock(Runnable.class);
+ onSucess = mock(Runnable.class);
+
+ action1 = spy(
+ RuntimeUtils.Actions.Action.builder()
+ .actionName("action1")
+ .numRetries(10)
+ .sleepBetweenInvocationsMs(100)
+ .supplier(supplier1)
+ .continueOn(false)
+ .onFail(onFail)
+ .onSuccess(onSucess)
+ .build());
+
+ action2 = spy(
+ RuntimeUtils.Actions.Action.builder()
+ .actionName("action2")
+ .numRetries(20)
+ .sleepBetweenInvocationsMs(200)
+ .supplier(supplier2)
+ .onFail(onFail)
+ .onSuccess(onSucess)
+ .build());
+
+ actions = RuntimeUtils.Actions.newBuilder()
+ .addAction(action1)
+ .addAction(action2);
+ actions.run();
+
+ Assert.assertEquals(actions.numActions(), 2);
+ verify(supplier1, times(1)).get();
+ verify(onFail, times(0)).run();
+ verify(onSucess, times(1)).run();
+ verify(supplier2, times(0)).get();
+
+ // test retry
+
+ supplier1 = mock(Supplier.class);
+ when(supplier1.get()).thenReturn(RuntimeUtils.Actions.ActionResult.builder().success(false).build());
+
+ supplier2 = mock(Supplier.class);
+ when(supplier2.get()).thenReturn(RuntimeUtils.Actions.ActionResult.builder().success(true).build());
+
+ onFail = mock(Runnable.class);
+ onSucess = mock(Runnable.class);
+
+ action1 = spy(
+ RuntimeUtils.Actions.Action.builder()
+ .actionName("action1")
+ .numRetries(10)
+ .sleepBetweenInvocationsMs(10)
+ .supplier(supplier1)
+ .continueOn(false)
+ .onFail(onFail)
+ .onSuccess(onSucess)
+ .build());
+
+ action2 = spy(
+ RuntimeUtils.Actions.Action.builder()
+ .actionName("action2")
+ .numRetries(20)
+ .sleepBetweenInvocationsMs(200)
+ .supplier(supplier2)
+ .build());
+
+ actions = RuntimeUtils.Actions.newBuilder()
+ .addAction(action1)
+ .addAction(action2);
+ actions.run();
+
+ Assert.assertEquals(actions.numActions(), 2);
+ verify(supplier1, times(10)).get();
+ verify(onFail, times(1)).run();
+ verify(onSucess, times(0)).run();
+ verify(supplier2, times(1)).get();
+
+ }
+}
diff --git a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/FunctionActioner.java b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/FunctionActioner.java
index 0c96864..8033507 100644
--- a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/FunctionActioner.java
+++ b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/FunctionActioner.java
@@ -18,45 +18,23 @@
*/
package org.apache.pulsar.functions.worker;
-import static org.apache.commons.lang3.StringUtils.isBlank;
-import static org.apache.pulsar.common.functions.Utils.FILE;
-import static org.apache.pulsar.common.functions.Utils.HTTP;
-import static org.apache.pulsar.functions.utils.Utils.getSourceType;
-import static org.apache.pulsar.functions.utils.Utils.getSinkType;
-import static org.apache.pulsar.common.functions.Utils.isFunctionPackageUrlSupported;
-
import com.google.common.annotations.VisibleForTesting;
import com.google.common.io.MoreFiles;
import com.google.common.io.RecursiveDeleteOption;
-
-import java.io.File;
-import java.io.FileNotFoundException;
-import java.io.FileOutputStream;
-import java.io.IOException;
-import java.net.URL;
-import java.nio.file.FileAlreadyExistsException;
-import java.nio.file.Files;
-import java.nio.file.Paths;
-import java.util.Collections;
-import java.util.Map;
-import java.util.UUID;
-import java.util.concurrent.LinkedBlockingQueue;
-import java.util.concurrent.TimeUnit;
-import java.util.function.Consumer;
-
import lombok.Data;
import lombok.EqualsAndHashCode;
import lombok.Getter;
import lombok.Setter;
import lombok.ToString;
import lombok.extern.slf4j.Slf4j;
-
import org.apache.commons.lang3.StringUtils;
import org.apache.distributedlog.api.namespace.Namespace;
import org.apache.pulsar.client.admin.PulsarAdmin;
import org.apache.pulsar.client.admin.PulsarAdminException;
import org.apache.pulsar.common.naming.TopicName;
import org.apache.pulsar.common.nar.NarClassLoader;
+import org.apache.pulsar.common.policies.data.SubscriptionStats;
+import org.apache.pulsar.common.policies.data.TopicStats;
import org.apache.pulsar.functions.instance.InstanceConfig;
import org.apache.pulsar.functions.instance.InstanceUtils;
import org.apache.pulsar.functions.proto.Function;
@@ -67,9 +45,34 @@ import org.apache.pulsar.functions.proto.Function.SinkSpec;
import org.apache.pulsar.functions.proto.Function.SourceSpec;
import org.apache.pulsar.functions.runtime.RuntimeFactory;
import org.apache.pulsar.functions.runtime.RuntimeSpawner;
+import org.apache.pulsar.functions.runtime.RuntimeUtils;
import org.apache.pulsar.functions.utils.FunctionDetailsUtils;
import org.apache.pulsar.functions.utils.io.ConnectorUtils;
+import java.io.File;
+import java.io.FileNotFoundException;
+import java.io.FileOutputStream;
+import java.io.IOException;
+import java.net.URL;
+import java.nio.file.FileAlreadyExistsException;
+import java.nio.file.Files;
+import java.nio.file.Paths;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+import java.util.function.Consumer;
+import java.util.stream.Collectors;
+
+import static org.apache.commons.lang3.StringUtils.isBlank;
+import static org.apache.pulsar.common.functions.Utils.FILE;
+import static org.apache.pulsar.common.functions.Utils.HTTP;
+import static org.apache.pulsar.common.functions.Utils.isFunctionPackageUrlSupported;
+import static org.apache.pulsar.functions.utils.Utils.getSinkType;
+import static org.apache.pulsar.functions.utils.Utils.getSourceType;
+
@Data
@Setter
@Getter
@@ -304,10 +307,10 @@ public class FunctionActioner implements AutoCloseable {
}
private void terminateFunction(FunctionRuntimeInfo functionRuntimeInfo) {
- FunctionDetails details = functionRuntimeInfo.getFunctionInstance().getFunctionMetaData()
- .getFunctionDetails();
- log.info("{}/{}/{}-{} Terminating function...", details.getTenant(), details.getNamespace(), details.getName(),
- functionRuntimeInfo.getFunctionInstance().getInstanceId());
+ FunctionDetails details = functionRuntimeInfo.getFunctionInstance().getFunctionMetaData().getFunctionDetails();
+ String fqfn = FunctionDetailsUtils.getFullyQualifiedName(details);
+
+ log.info("{}-{} Terminating function...", fqfn,functionRuntimeInfo.getFunctionInstance().getInstanceId());
stopFunction(functionRuntimeInfo);
//cleanup subscriptions
@@ -319,27 +322,64 @@ public class FunctionActioner implements AutoCloseable {
Function.ConsumerSpec consumerSpec = stringConsumerSpecEntry.getValue();
String topic = stringConsumerSpecEntry.getKey();
- String subscriptionName = functionRuntimeInfo
- .getFunctionInstance().getFunctionMetaData()
- .getFunctionDetails().getSource().getSubscriptionName();
- // if user specified subscription name is empty use default subscription name
- if (isBlank(subscriptionName)) {
- subscriptionName = InstanceUtils.getDefaultSubscriptionName(
- functionRuntimeInfo.getFunctionInstance()
- .getFunctionMetaData().getFunctionDetails());
- }
+
+ String subscriptionName = isBlank(functionRuntimeInfo.getFunctionInstance().getFunctionMetaData().getFunctionDetails().getSource().getSubscriptionName())
+ ? InstanceUtils.getDefaultSubscriptionName(functionRuntimeInfo.getFunctionInstance().getFunctionMetaData().getFunctionDetails())
+ : functionRuntimeInfo.getFunctionInstance().getFunctionMetaData().getFunctionDetails().getSource().getSubscriptionName();
try {
- if (consumerSpec.getIsRegexPattern()) {
- pulsarAdmin.namespaces().unsubscribeNamespace(TopicName.get(topic).getNamespace(), subscriptionName);
- } else {
- pulsarAdmin.topics().deleteSubscription(topic, subscriptionName);
- }
- } catch (PulsarAdminException e) {
- log.warn("Failed to cleanup {} subscription for {}", subscriptionName,
- FunctionDetailsUtils.getFullyQualifiedName(
- functionRuntimeInfo.getFunctionInstance()
- .getFunctionMetaData().getFunctionDetails()), e);
+ RuntimeUtils.Actions.newBuilder()
+ .addAction(
+ RuntimeUtils.Actions.Action.builder()
+ .actionName(String.format("Cleaning up subscriptions for function %s", fqfn))
+ .numRetries(10)
+ .sleepBetweenInvocationsMs(1000)
+ .supplier(() -> {
+ try {
+ if (consumerSpec.getIsRegexPattern()) {
+ pulsarAdmin.namespaces().unsubscribeNamespace(TopicName
+ .get(topic).getNamespace(), subscriptionName);
+ } else {
+ pulsarAdmin.topics().deleteSubscription(topic,
+ subscriptionName);
+ }
+ } catch (PulsarAdminException e) {
+ if (e instanceof PulsarAdminException.NotFoundException) {
+ return RuntimeUtils.Actions.ActionResult.builder()
+ .success(true)
+ .build();
+ } else {
+ // for debugging purposes
+ List<Map<String, String>> existingConsumers = Collections.emptyList();
+ try {
+ TopicStats stats = pulsarAdmin.topics().getStats(topic);
+ SubscriptionStats sub = stats.subscriptions.get(InstanceUtils.getDefaultSubscriptionName(details));
+ if (sub != null) {
+ existingConsumers = sub.consumers.stream()
+ .map(consumerStats -> consumerStats.metadata)
+ .collect(Collectors.toList());
+ }
+ } catch (PulsarAdminException e1) {
+
+ }
+
+ String errorMsg = e.getHttpError() != null ? e.getHttpError() : e.getMessage();
+ return RuntimeUtils.Actions.ActionResult.builder()
+ .success(false)
+ .errorMsg(String.format("%s - existing consumers: %s", errorMsg, existingConsumers))
+ .build();
+ }
+ }
+
+ return RuntimeUtils.Actions.ActionResult.builder()
+ .success(true)
+ .build();
+
+ })
+ .build())
+ .run();
+ } catch (InterruptedException e) {
+ throw new RuntimeException(e);
}
}
});