existingFunctionAuthData = Optional.empty();
diff --git a/pulsar-functions/runtime/src/test/java/org/apache/pulsar/functions/runtime/KubernetesRuntimeFactoryTest.java b/pulsar-functions/runtime/src/test/java/org/apache/pulsar/functions/runtime/KubernetesRuntimeFactoryTest.java
index 5ea75fa296fe1..37d6c1323ef10 100644
--- a/pulsar-functions/runtime/src/test/java/org/apache/pulsar/functions/runtime/KubernetesRuntimeFactoryTest.java
+++ b/pulsar-functions/runtime/src/test/java/org/apache/pulsar/functions/runtime/KubernetesRuntimeFactoryTest.java
@@ -59,10 +59,12 @@ public String getSecretsProviderClassName(FunctionDetails functionDetails) {
if (!StringUtils.isEmpty(functionDetails.getSecretsMap())) {
if (functionDetails.getRuntime() == FunctionDetails.Runtime.JAVA) {
return ClearTextSecretsProvider.class.getName();
- } else {
+ }
+ else {
return "secretsprovider.ClearTextSecretsProvider";
}
- } else {
+ }
+ else {
return null;
}
}
@@ -125,27 +127,31 @@ public void tearDown() {
KubernetesRuntimeFactory createKubernetesRuntimeFactory(String extraDepsDir, Resources minResources) throws Exception {
KubernetesRuntimeFactory factory = spy(new KubernetesRuntimeFactory(
- null,
- null,
- null,
- null,
- pulsarRootDir,
- false,
- true,
- "myrepo",
- "anotherrepo",
- extraDepsDir,
- null,
+ null,
+ null,
+ null,
+ null,
+ pulsarRootDir,
+ false,
+ true,
+ "myrepo",
+ "anotherrepo",
+ extraDepsDir,
+ null,
0,
pulsarServiceUrl,
- pulsarAdminUrl,
- stateStorageServiceUrl,
- null,
- null,
- null,
- null,
+ pulsarAdminUrl,
+ stateStorageServiceUrl,
+ null,
+ null,
+ null,
+ null,
+ null,
minResources,
- new TestSecretProviderConfigurator(), false));
+ new TestSecretProviderConfigurator(),
+ false,
+ "",
+ ""));
doNothing().when(factory).setupClient();
return factory;
}
@@ -182,8 +188,8 @@ public void testValidateMinResourcesRequired() throws Exception {
testMinResource(0.2, 2048L, false, null);
testMinResource(0.05, 2048L, true, "Per instance CPU requested, 0.05, for function is less than the minimum required, 0.1");
- testMinResource(0.2,512L, true, "Per instance RAM requested, 512, for function is less than the minimum required, 1024");
- testMinResource(0.05,512L, true, "Per instance CPU requested, 0.05, for function is less than the minimum required, 0.1");
+ testMinResource(0.2, 512L, true, "Per instance RAM requested, 512, for function is less than the minimum required, 1024");
+ testMinResource(0.05, 512L, true, "Per instance CPU requested, 0.05, for function is less than the minimum required, 0.1");
testMinResource(null, null, true, "Per instance CPU requested, 0.0, for function is less than the minimum required, 0.1");
testMinResource(0.2, null, true, "Per instance RAM requested, 0, for function is less than the minimum required, 1024");
@@ -207,7 +213,8 @@ private void testMinResource(Double cpu, Long ram, boolean fail, String failErro
FunctionDetails functionDetails;
if (ram != null || cpu != null) {
functionDetails = FunctionDetails.newBuilder(functionDetailsBase).setResources(resources).build();
- } else {
+ }
+ else {
functionDetails = FunctionDetails.newBuilder(functionDetailsBase).build();
}
diff --git a/pulsar-functions/runtime/src/test/java/org/apache/pulsar/functions/runtime/KubernetesRuntimeTest.java b/pulsar-functions/runtime/src/test/java/org/apache/pulsar/functions/runtime/KubernetesRuntimeTest.java
index 12390dcaf1481..42d44cb1fbd8c 100644
--- a/pulsar-functions/runtime/src/test/java/org/apache/pulsar/functions/runtime/KubernetesRuntimeTest.java
+++ b/pulsar-functions/runtime/src/test/java/org/apache/pulsar/functions/runtime/KubernetesRuntimeTest.java
@@ -6,9 +6,9 @@
* 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
- *
+ *
+ * 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
@@ -19,11 +19,12 @@
package org.apache.pulsar.functions.runtime;
+import com.google.gson.JsonArray;
+import com.google.gson.JsonObject;
import com.google.protobuf.util.JsonFormat;
import io.kubernetes.client.apis.AppsV1Api;
import io.kubernetes.client.apis.CoreV1Api;
-import io.kubernetes.client.models.V1Container;
-import io.kubernetes.client.models.V1PodSpec;
+import io.kubernetes.client.models.*;
import org.apache.commons.lang.StringUtils;
import org.apache.pulsar.functions.instance.InstanceConfig;
import org.apache.pulsar.functions.proto.Function;
@@ -80,10 +81,12 @@ public String getSecretsProviderClassName(FunctionDetails functionDetails) {
if (!StringUtils.isEmpty(functionDetails.getSecretsMap())) {
if (functionDetails.getRuntime() == FunctionDetails.Runtime.JAVA) {
return ClearTextSecretsProvider.class.getName();
- } else {
+ }
+ else {
return "secretsprovider.ClearTextSecretsProvider";
}
- } else {
+ }
+ else {
return null;
}
}
@@ -149,33 +152,46 @@ public void tearDown() {
}
}
- KubernetesRuntimeFactory createKubernetesRuntimeFactory(String extraDepsDir, int percentMemoryPadding) throws Exception {
+ KubernetesRuntimeFactory createKubernetesRuntimeFactory(String extraDepsDir, int percentMemoryPadding, String manifestCustomizerClassName) throws Exception {
KubernetesRuntimeFactory factory = spy(new KubernetesRuntimeFactory(
- null,
- null,
- null,
- null,
- pulsarRootDir,
- false,
- true,
- "myrepo",
- "anotherrepo",
- extraDepsDir,
- null,
+ null,
+ null,
+ null,
+ null,
+ pulsarRootDir,
+ false,
+ true,
+ "myrepo",
+ "anotherrepo",
+ extraDepsDir,
+ null,
percentMemoryPadding,
pulsarServiceUrl,
- pulsarAdminUrl,
- stateStorageServiceUrl,
- null,
- null,
- null,
- null,
- null, new TestSecretProviderConfigurator(), false));
+ pulsarAdminUrl,
+ stateStorageServiceUrl,
+ null,
+ null,
+ null,
+ null,
+ null,
+ null,
+ new TestSecretProviderConfigurator(),
+ false,
+ manifestCustomizerClassName,
+ ""));
doNothing().when(factory).setupClient();
return factory;
+
+ }
+ KubernetesRuntimeFactory createKubernetesRuntimeFactory(String extraDepsDir, int percentMemoryPadding) throws Exception {
+ return createKubernetesRuntimeFactory(extraDepsDir, percentMemoryPadding, null);
}
FunctionDetails createFunctionDetails(FunctionDetails.Runtime runtime, boolean addSecrets) {
+ return createFunctionDetails(runtime, addSecrets, (fb) -> fb);
+ }
+
+ FunctionDetails createFunctionDetails(FunctionDetails.Runtime runtime, boolean addSecrets, java.util.function.Function customize) {
FunctionDetails.Builder functionDetailsBuilder = FunctionDetails.newBuilder();
functionDetailsBuilder.setRuntime(runtime);
functionDetailsBuilder.setTenant(TEST_TENANT);
@@ -198,7 +214,7 @@ FunctionDetails createFunctionDetails(FunctionDetails.Runtime runtime, boolean a
functionDetailsBuilder.setSecretsMap("SomeMap");
}
functionDetailsBuilder.setResources(RESOURCES);
- return functionDetailsBuilder.build();
+ return customize.apply(functionDetailsBuilder).build();
}
InstanceConfig createJavaInstanceConfig(FunctionDetails.Runtime runtime, boolean addSecrets) {
@@ -235,7 +251,6 @@ private void verifyRamPadding(int percentMemoryPadding, long ram, long expectedR
}
-
@Test
public void testJavaConstructor() throws Exception {
InstanceConfig config = createJavaInstanceConfig(FunctionDetails.Runtime.JAVA, false);
@@ -281,7 +296,8 @@ private void verifyJavaInstance(InstanceConfig config, String depsDir, boolean s
totalArgs = 35;
portArg = 26;
metricsPortArg = 28;
- } else {
+ }
+ else {
extraDepsEnv = "";
portArg = 25;
metricsPortArg = 27;
@@ -292,7 +308,7 @@ private void verifyJavaInstance(InstanceConfig config, String depsDir, boolean s
}
assertEquals(args.size(), totalArgs,
- "Actual args : " + StringUtils.join(args, " "));
+ "Actual args : " + StringUtils.join(args, " "));
String expectedArgs = "exec java -cp " + classpath
+ extraDepsEnv
@@ -364,7 +380,8 @@ private void verifyPythonInstance(InstanceConfig config, String extraDepsDir, bo
configArg = 10;
pythonPath = "";
metricsPortArg = 32;
- } else {
+ }
+ else {
totalArgs = 40;
portArg = 31;
configArg = 11;
@@ -376,7 +393,7 @@ private void verifyPythonInstance(InstanceConfig config, String extraDepsDir, bo
}
assertEquals(args.size(), totalArgs,
- "Actual args : " + StringUtils.join(args, " "));
+ "Actual args : " + StringUtils.join(args, " "));
String expectedArgs = pythonPath + "exec python " + pythonInstanceFile
+ " --py " + pulsarRootDir + "/" + userJarFile
+ " --logging_directory " + logDirectory
@@ -410,4 +427,70 @@ private void verifyPythonInstance(InstanceConfig config, String extraDepsDir, bo
assertEquals(containerSpec.getResources().getLimits().get("cpu").getNumber().doubleValue(), RESOURCES.getCpu());
}
+ @Test
+ public void testDefaultKubernetesManifestCustomizer() throws Exception {
+ InstanceConfig config = createJavaInstanceConfig(FunctionDetails.Runtime.JAVA, false);
+ config.setFunctionDetails(createFunctionDetails(FunctionDetails.Runtime.JAVA, false, (fb) -> {
+ JsonObject configObj = new JsonObject();
+ configObj.addProperty("jobNamespace", "custom-ns");
+
+ JsonObject extraAnn = new JsonObject();
+ extraAnn.addProperty("annotation", "test");
+ configObj.add("extraAnnotations", extraAnn);
+
+ JsonObject extraLabel = new JsonObject();
+ extraLabel.addProperty("label", "test");
+ configObj.add("extraLabels", extraLabel);
+
+ JsonObject nodeLabels = new JsonObject();
+ nodeLabels.addProperty("selector", "test");
+ configObj.add("nodeSelectorLabels", nodeLabels);
+
+ JsonArray tolerations = new JsonArray();
+ JsonObject toleration = new JsonObject();
+ toleration.addProperty("key", "test");
+ toleration.addProperty("value", "test");
+ toleration.addProperty("effect", "test");
+ tolerations.add(toleration);
+ configObj.add("tolerations", tolerations);
+
+ return fb.setCustomRuntimeOptions(configObj.toString());
+ }));
+
+ factory = createKubernetesRuntimeFactory(null, 10);
+
+ verifyJavaInstance(config, pulsarRootDir + "/instances/deps", false);
+ KubernetesRuntime container = factory.createContainer(config, userJarFile, userJarFile, 30l);
+ V1StatefulSet spec = container.createStatefulSet();
+ assertEquals(spec.getMetadata().getAnnotations().get("annotation"), "test");
+ assertEquals(spec.getMetadata().getLabels().get("label"), "test");
+ assertEquals(spec.getSpec().getTemplate().getSpec().getNodeSelector().get("selector"), "test");
+ List tols = spec.getSpec().getTemplate().getSpec().getTolerations();
+ // we add three by default, plus our custom
+ assertEquals(tols.size(), 4);
+ assertEquals(tols.get(3).getKey(), "test");
+ assertEquals(tols.get(3).getValue(), "test");
+ assertEquals(tols.get(3).getEffect(), "test");
+
+ V1Service serviceSpec = container.createService();
+ assertEquals(serviceSpec.getMetadata().getNamespace(), "custom-ns");
+ assertEquals(serviceSpec.getMetadata().getAnnotations().get("annotation"), "test");
+ assertEquals(serviceSpec.getMetadata().getLabels().get("label"), "test");
+
+ }
+
+ @Test
+ public void testCustomKubernetesManifestCustomizer() throws Exception {
+ InstanceConfig config = createJavaInstanceConfig(FunctionDetails.Runtime.JAVA, false);
+ config.setFunctionDetails(createFunctionDetails(FunctionDetails.Runtime.JAVA, false, (fb) -> {
+ return fb.setCustomRuntimeOptions("custom-service-account");
+ }));
+
+ factory = createKubernetesRuntimeFactory(null, 10, TestKubernetesCustomManifestCustomizer.class.getCanonicalName());
+
+ verifyJavaInstance(config, pulsarRootDir + "/instances/deps", false);
+ KubernetesRuntime container = factory.createContainer(config, userJarFile, userJarFile, 30l);
+ V1StatefulSet spec = container.createStatefulSet();
+ assertEquals(spec.getSpec().getTemplate().getSpec().getServiceAccountName(), "my-service-account");
+ }
}
diff --git a/pulsar-functions/runtime/src/test/java/org/apache/pulsar/functions/runtime/TestKubernetesCustomManifestCustomizer.java b/pulsar-functions/runtime/src/test/java/org/apache/pulsar/functions/runtime/TestKubernetesCustomManifestCustomizer.java
new file mode 100644
index 0000000000000..904ae9e63e83b
--- /dev/null
+++ b/pulsar-functions/runtime/src/test/java/org/apache/pulsar/functions/runtime/TestKubernetesCustomManifestCustomizer.java
@@ -0,0 +1,19 @@
+package org.apache.pulsar.functions.runtime;
+
+import io.kubernetes.client.models.V1StatefulSet;
+import org.apache.pulsar.functions.proto.Function;
+
+import static org.testng.Assert.assertEquals;
+
+class TestKubernetesCustomManifestCustomizer implements KubernetesManifestCustomizer {
+ @Override
+ public void init(Function.FunctionDetails funcDetails, String staticConfig) {
+ assertEquals(funcDetails.getCustomRuntimeOptions(), "custom-service-account");
+ }
+
+ @Override
+ public V1StatefulSet customizeStatefulSet(V1StatefulSet statefulSet) {
+ statefulSet.getSpec().getTemplate().getSpec().serviceAccountName("my-service-account");
+ return statefulSet;
+ }
+}
diff --git a/pulsar-functions/secrets/pom.xml b/pulsar-functions/secrets/pom.xml
index 7dd8cbd4b71b2..164fd923c9fef 100644
--- a/pulsar-functions/secrets/pom.xml
+++ b/pulsar-functions/secrets/pom.xml
@@ -24,7 +24,7 @@
org.apache.pulsar
pulsar-functions
- 2.4.1-inst1
+ 2.4.1-inst3
pulsar-functions-secrets
diff --git a/pulsar-functions/utils/pom.xml b/pulsar-functions/utils/pom.xml
index 2ec03f88a937b..9b5fa7bacc0df 100644
--- a/pulsar-functions/utils/pom.xml
+++ b/pulsar-functions/utils/pom.xml
@@ -25,7 +25,7 @@
org.apache.pulsar
pulsar-functions
- 2.4.1-inst1
+ 2.4.1-inst3
pulsar-functions-utils
diff --git a/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/FunctionConfigUtils.java b/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/FunctionConfigUtils.java
index 43e034682a57f..43ea0daa3f15a 100644
--- a/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/FunctionConfigUtils.java
+++ b/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/FunctionConfigUtils.java
@@ -234,6 +234,10 @@ public static FunctionDetails convert(FunctionConfig functionConfig, ClassLoader
functionDetailsBuilder.setComponentType(FunctionDetails.ComponentType.FUNCTION);
+ if (!StringUtils.isEmpty(functionConfig.getCustomRuntimeOptions())) {
+ functionDetailsBuilder.setCustomRuntimeOptions(functionConfig.getCustomRuntimeOptions());
+ }
+
return functionDetailsBuilder.build();
}
@@ -334,6 +338,10 @@ public static FunctionConfig convertFromDetails(FunctionDetails functionDetails)
functionConfig.setRuntimeFlags(functionDetails.getRuntimeFlags());
}
+ if (!isEmpty(functionDetails.getCustomRuntimeOptions())) {
+ functionConfig.setCustomRuntimeOptions(functionDetails.getCustomRuntimeOptions());
+ }
+
return functionConfig;
}
@@ -775,6 +783,9 @@ public static FunctionConfig validateUpdate(FunctionConfig existingConfig, Funct
if (!StringUtils.isEmpty(newConfig.getRuntimeFlags())) {
mergedConfig.setRuntimeFlags(newConfig.getRuntimeFlags());
}
+ if (!StringUtils.isEmpty(newConfig.getCustomRuntimeOptions())) {
+ mergedConfig.setCustomRuntimeOptions(newConfig.getCustomRuntimeOptions());
+ }
return mergedConfig;
}
}
diff --git a/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/SinkConfigUtils.java b/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/SinkConfigUtils.java
index f0a39b93ca864..800a55f560348 100644
--- a/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/SinkConfigUtils.java
+++ b/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/SinkConfigUtils.java
@@ -209,6 +209,10 @@ public static FunctionDetails convert(SinkConfig sinkConfig, ExtractedSinkDetail
functionDetailsBuilder.setComponentType(FunctionDetails.ComponentType.SINK);
+ if (!StringUtils.isEmpty(sinkConfig.getCustomRuntimeOptions())) {
+ functionDetailsBuilder.setCustomRuntimeOptions(sinkConfig.getCustomRuntimeOptions());
+ }
+
return functionDetailsBuilder.build();
}
@@ -283,6 +287,10 @@ public static SinkConfig convertFromDetails(FunctionDetails functionDetails) {
sinkConfig.setRuntimeFlags(functionDetails.getRuntimeFlags());
}
+ if (!isEmpty(functionDetails.getCustomRuntimeOptions())) {
+ sinkConfig.setCustomRuntimeOptions(functionDetails.getCustomRuntimeOptions());
+ }
+
return sinkConfig;
}
@@ -560,6 +568,9 @@ public static SinkConfig validateUpdate(SinkConfig existingConfig, SinkConfig ne
if (!StringUtils.isEmpty(newConfig.getRuntimeFlags())) {
mergedConfig.setRuntimeFlags(newConfig.getRuntimeFlags());
}
+ if (!StringUtils.isEmpty(newConfig.getCustomRuntimeOptions())) {
+ mergedConfig.setCustomRuntimeOptions(newConfig.getCustomRuntimeOptions());
+ }
return mergedConfig;
}
}
\ No newline at end of file
diff --git a/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/SourceConfigUtils.java b/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/SourceConfigUtils.java
index 149204cf5dc67..0c51b60cbfb34 100644
--- a/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/SourceConfigUtils.java
+++ b/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/SourceConfigUtils.java
@@ -144,6 +144,10 @@ public static FunctionDetails convert(SourceConfig sourceConfig, ExtractedSource
functionDetailsBuilder.setComponentType(FunctionDetails.ComponentType.SOURCE);
+ if (!StringUtils.isEmpty(sourceConfig.getCustomRuntimeOptions())) {
+ functionDetailsBuilder.setCustomRuntimeOptions(sourceConfig.getCustomRuntimeOptions());
+ }
+
return functionDetailsBuilder.build();
}
@@ -194,9 +198,14 @@ public static SourceConfig convertFromDetails(FunctionDetails functionDetails) {
sourceConfig.setResources(resources);
}
- if (!org.apache.commons.lang3.StringUtils.isEmpty(functionDetails.getRuntimeFlags())) {
- sourceConfig .setRuntimeFlags(functionDetails.getRuntimeFlags());
+ if (!isEmpty(functionDetails.getRuntimeFlags())) {
+ sourceConfig.setRuntimeFlags(functionDetails.getRuntimeFlags());
+ }
+
+ if (!isEmpty(functionDetails.getCustomRuntimeOptions())) {
+ sourceConfig.setCustomRuntimeOptions(functionDetails.getCustomRuntimeOptions());
}
+
return sourceConfig;
}
@@ -371,6 +380,9 @@ public static SourceConfig validateUpdate(SourceConfig existingConfig, SourceCon
if (!StringUtils.isEmpty(newConfig.getRuntimeFlags())) {
mergedConfig.setRuntimeFlags(newConfig.getRuntimeFlags());
}
+ if (!StringUtils.isEmpty(newConfig.getCustomRuntimeOptions())) {
+ mergedConfig.setCustomRuntimeOptions(newConfig.getCustomRuntimeOptions());
+ }
return mergedConfig;
}
diff --git a/pulsar-functions/worker/pom.xml b/pulsar-functions/worker/pom.xml
index 53a1a787c78ab..bbc2960972747 100644
--- a/pulsar-functions/worker/pom.xml
+++ b/pulsar-functions/worker/pom.xml
@@ -25,7 +25,7 @@
org.apache.pulsar
pulsar-functions
- 2.4.1-inst1
+ 2.4.1-inst3
..
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 7262007697f06..8557a7dc669f9 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
@@ -290,6 +290,7 @@ public void terminateFunction(FunctionRuntimeInfo functionRuntimeInfo) {
FunctionDetails details = functionRuntimeInfo.getFunctionInstance().getFunctionMetaData().getFunctionDetails();
String fqfn = FunctionCommon.getFullyQualifiedName(details);
log.info("{}-{} Terminating function...", fqfn,functionRuntimeInfo.getFunctionInstance().getInstanceId());
+ FunctionDetails funcDetails = functionRuntimeInfo.getFunctionInstance().getFunctionMetaData().getFunctionDetails();
if (functionRuntimeInfo.getRuntimeSpawner() != null) {
functionRuntimeInfo.getRuntimeSpawner().close();
@@ -299,7 +300,7 @@ public void terminateFunction(FunctionRuntimeInfo functionRuntimeInfo) {
try {
log.info("{}-{} Cleaning up authentication data for function...", fqfn,functionRuntimeInfo.getFunctionInstance().getInstanceId());
functionRuntimeInfo.getRuntimeSpawner()
- .getRuntimeFactory().getAuthProvider()
+ .getRuntimeFactory().getAuthProvider(funcDetails)
.cleanUpAuthData(
details.getTenant(), details.getNamespace(), details.getName(),
Optional.ofNullable(getFunctionAuthData(
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
index 706c5a9ad1ae2..19aa3737bd18e 100644
--- 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
@@ -34,11 +34,7 @@
import org.apache.pulsar.functions.instance.AuthenticationConfig;
import org.apache.pulsar.functions.proto.Function;
import org.apache.pulsar.functions.proto.Function.Assignment;
-import org.apache.pulsar.functions.runtime.KubernetesRuntimeFactory;
-import org.apache.pulsar.functions.runtime.ProcessRuntimeFactory;
-import org.apache.pulsar.functions.runtime.RuntimeFactory;
-import org.apache.pulsar.functions.runtime.RuntimeSpawner;
-import org.apache.pulsar.functions.runtime.ThreadRuntimeFactory;
+import org.apache.pulsar.functions.runtime.*;
import org.apache.pulsar.functions.secretsprovider.ClearTextSecretsProvider;
import org.apache.pulsar.functions.secretsproviderconfigurator.DefaultSecretsProviderConfigurator;
import org.apache.pulsar.functions.secretsproviderconfigurator.SecretsProviderConfigurator;
@@ -170,12 +166,15 @@ public FunctionRuntimeManager(WorkerConfig workerConfig, WorkerService workerSer
StringUtils.isEmpty(workerConfig.getKubernetesContainerFactory().getPulsarAdminUrl()) ? workerConfig.getPulsarWebServiceUrl() : workerConfig.getKubernetesContainerFactory().getPulsarAdminUrl(),
workerConfig.getStateStorageServiceUrl(),
authConfig,
+ workerConfig.getTlsTrustChainBytes(),
workerConfig.getKubernetesContainerFactory().getExpectedMetricsCollectionInterval() == null ? -1 : workerConfig.getKubernetesContainerFactory().getExpectedMetricsCollectionInterval(),
workerConfig.getKubernetesContainerFactory().getChangeConfigMap(),
workerConfig.getKubernetesContainerFactory().getChangeConfigMapNamespace(),
workerConfig.getFunctionInstanceMinResources(),
secretsProviderConfigurator,
- workerConfig.isAuthenticationEnabled());
+ workerConfig.isAuthenticationEnabled(),
+ workerConfig.getKubernetesContainerFactory().getKubernetesManifestCustomizerClassName(),
+ workerConfig.getKubernetesContainerFactory().getKubernetesManifestCustomizerConfig());
} else {
throw new RuntimeException("Either Thread, Process or Kubernetes Container Factory need to be set");
}
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
index 79a944898bd78..1e957bb01139c 100644
--- 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
@@ -28,6 +28,8 @@
import java.io.Serializable;
import java.net.InetAddress;
import java.net.UnknownHostException;
+import java.nio.file.Files;
+import java.nio.file.Paths;
import java.util.Map;
import java.util.Properties;
import java.util.Set;
@@ -386,8 +388,8 @@ public static class KubernetesContainerFactory {
)
private String k8Uri;
@FieldContext(
- doc = "The Kubernetes namespace to run the function instances. It is `default`,"
- + " if this setting is left to be empty"
+ doc = "The default Kubernetes namespace to run the function instances. It is `default`,"
+ + " if this setting is left to be empty. May be overridden by a KubernetesManifestCustomizer."
)
private String jobNamespace;
@FieldContext(
@@ -464,6 +466,20 @@ public static class KubernetesContainerFactory {
doc = "Additional memory padding added on top of the memory requested by the function per on a per instance basis"
)
private int percentMemoryPadding;
+
+ @FieldContext(
+ doc = "The full class-name of an instance of KubernetesManifestCustomizer." +
+ " This class receives the 'customRuntimeOptions string and can customize" +
+ " the generation of kubernetes manifest files"
+ )
+ private String kubernetesManifestCustomizerClassName;
+
+ @FieldContext(
+ doc = "Config that can be passed to the KubernetesManifestCustomizer." +
+ " This config is distinct from the `customRuntimeOptions` provided by functions" +
+ " as this config is the the same across all functions and is static "
+ )
+ private String kubernetesManifestCustomizerConfig = "";
}
@FieldContext(
category = CATEGORY_FUNC_RUNTIME_MNG,
@@ -519,6 +535,18 @@ public String getWorkerHostname() {
return this.workerHostname;
}
+ public byte[] getTlsTrustChainBytes() {
+ if (StringUtils.isNotEmpty(getTlsTrustCertsFilePath()) && Files.exists(Paths.get(getTlsTrustCertsFilePath()))) {
+ try {
+ return Files.readAllBytes(Paths.get(getTlsTrustCertsFilePath()));
+ } catch (IOException e) {
+ throw new IllegalStateException("Failed to read CA bytes", e);
+ }
+ } else {
+ return null;
+ }
+ }
+
public String getWorkerWebAddress() {
return String.format("http://%s:%d", this.getWorkerHostname(), this.getWorkerPort());
}
diff --git a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/FunctionsImpl.java b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/FunctionsImpl.java
index d96ed30136741..527352b1e0ead 100644
--- a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/FunctionsImpl.java
+++ b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/FunctionsImpl.java
@@ -191,7 +191,7 @@ public void registerFunction(final String tenant,
try {
Optional functionAuthData = worker().getFunctionRuntimeManager()
.getRuntimeFactory()
- .getAuthProvider()
+ .getAuthProvider(functionDetails)
.cacheAuthData(tenant, namespace, functionName, clientAuthenticationDataHttps);
if (functionAuthData.isPresent()) {
@@ -373,7 +373,7 @@ public void updateFunction(final String tenant,
try {
Optional newFunctionAuthData = worker().getFunctionRuntimeManager()
.getRuntimeFactory()
- .getAuthProvider()
+ .getAuthProvider(functionDetails)
.updateAuthData(
tenant, namespace,
functionName, existingFunctionAuthData,
diff --git a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/SinksImpl.java b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/SinksImpl.java
index 659cb660e9ba4..b87f1291811ba 100644
--- a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/SinksImpl.java
+++ b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/SinksImpl.java
@@ -193,7 +193,7 @@ public void registerSink(final String tenant,
try {
Optional functionAuthData = worker().getFunctionRuntimeManager()
.getRuntimeFactory()
- .getAuthProvider()
+ .getAuthProvider(functionDetails)
.cacheAuthData(tenant, namespace, sinkName, clientAuthenticationDataHttps);
if (functionAuthData.isPresent()) {
@@ -378,7 +378,7 @@ public void updateSink(final String tenant,
try {
Optional newFunctionAuthData = worker().getFunctionRuntimeManager()
.getRuntimeFactory()
- .getAuthProvider()
+ .getAuthProvider(functionDetails)
.updateAuthData(
tenant, namespace,
sinkName, existingFunctionAuthData,
diff --git a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/SourcesImpl.java b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/SourcesImpl.java
index e9cfeb0dc8d39..e6eaffe0070f2 100644
--- a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/SourcesImpl.java
+++ b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/SourcesImpl.java
@@ -193,7 +193,7 @@ public void registerSource(final String tenant,
try {
Optional functionAuthData = worker().getFunctionRuntimeManager()
.getRuntimeFactory()
- .getAuthProvider()
+ .getAuthProvider(functionDetails)
.cacheAuthData(tenant, namespace, sourceName, clientAuthenticationDataHttps);
if (functionAuthData.isPresent()) {
@@ -375,7 +375,7 @@ public void updateSource(final String tenant,
try {
Optional newFunctionAuthData = worker().getFunctionRuntimeManager()
.getRuntimeFactory()
- .getAuthProvider()
+ .getAuthProvider(functionDetails)
.updateAuthData(
tenant, namespace,
sourceName, existingFunctionAuthData,
diff --git a/pulsar-functions/worker/src/test/java/org/apache/pulsar/functions/worker/FunctionActionerTest.java b/pulsar-functions/worker/src/test/java/org/apache/pulsar/functions/worker/FunctionActionerTest.java
index ca21c8b623f89..6cb5d99ef6ffe 100644
--- a/pulsar-functions/worker/src/test/java/org/apache/pulsar/functions/worker/FunctionActionerTest.java
+++ b/pulsar-functions/worker/src/test/java/org/apache/pulsar/functions/worker/FunctionActionerTest.java
@@ -193,7 +193,7 @@ public void testFunctionAuthDisabled() throws Exception {
RuntimeFactory runtimeFactory = mock(RuntimeFactory.class);
FunctionAuthProvider functionAuthProvider = mock(FunctionAuthProvider.class);
- doReturn(functionAuthProvider).when(runtimeFactory).getAuthProvider();
+ doReturn(functionAuthProvider).when(runtimeFactory).getAuthProvider(functionMeta.getFunctionDetails());
doReturn(runtimeFactory).when(runtimeSpawner).getRuntimeFactory();
doReturn(instance).when(functionRuntimeInfo).getFunctionInstance();
diff --git a/pulsar-io/aerospike/pom.xml b/pulsar-io/aerospike/pom.xml
index c89ee0adc854d..81e77e1539339 100644
--- a/pulsar-io/aerospike/pom.xml
+++ b/pulsar-io/aerospike/pom.xml
@@ -24,7 +24,7 @@
org.apache.pulsar
pulsar-io
- 2.4.1-inst1
+ 2.4.1-inst3
pulsar-io-aerospike
diff --git a/pulsar-io/canal/pom.xml b/pulsar-io/canal/pom.xml
index 84acbb9f86ae8..6325430778768 100644
--- a/pulsar-io/canal/pom.xml
+++ b/pulsar-io/canal/pom.xml
@@ -25,7 +25,7 @@
org.apache.pulsar
pulsar-io
- 2.4.1-inst1
+ 2.4.1-inst3
4.0.0
diff --git a/pulsar-io/cassandra/pom.xml b/pulsar-io/cassandra/pom.xml
index 62eba7b75f689..aae7a83a082c7 100644
--- a/pulsar-io/cassandra/pom.xml
+++ b/pulsar-io/cassandra/pom.xml
@@ -24,7 +24,7 @@
org.apache.pulsar
pulsar-io
- 2.4.1-inst1
+ 2.4.1-inst3
pulsar-io-cassandra
diff --git a/pulsar-io/common/pom.xml b/pulsar-io/common/pom.xml
index 0a90edcf39911..0f845e2f44dad 100644
--- a/pulsar-io/common/pom.xml
+++ b/pulsar-io/common/pom.xml
@@ -27,7 +27,7 @@
org.apache.pulsar
pulsar-io
- 2.4.1-inst1
+ 2.4.1-inst3
pulsar-io-common
diff --git a/pulsar-io/core/pom.xml b/pulsar-io/core/pom.xml
index e79e763167464..b5296579b9c7e 100644
--- a/pulsar-io/core/pom.xml
+++ b/pulsar-io/core/pom.xml
@@ -24,7 +24,7 @@
org.apache.pulsar
pulsar-io
- 2.4.1-inst1
+ 2.4.1-inst3
pulsar-io-core
diff --git a/pulsar-io/data-generator/pom.xml b/pulsar-io/data-generator/pom.xml
index d7e3d618e3f09..2a2ee0351e341 100644
--- a/pulsar-io/data-generator/pom.xml
+++ b/pulsar-io/data-generator/pom.xml
@@ -24,7 +24,7 @@
org.apache.pulsar
pulsar-io
- 2.4.1-inst1
+ 2.4.1-inst3
pulsar-io-data-generator
diff --git a/pulsar-io/debezium/core/pom.xml b/pulsar-io/debezium/core/pom.xml
index 78993117a88dd..8ec2669d31c62 100644
--- a/pulsar-io/debezium/core/pom.xml
+++ b/pulsar-io/debezium/core/pom.xml
@@ -24,7 +24,7 @@
org.apache.pulsar
pulsar-io-debezium
- 2.4.1-inst1
+ 2.4.1-inst3
pulsar-io-debezium-core
diff --git a/pulsar-io/debezium/mysql/pom.xml b/pulsar-io/debezium/mysql/pom.xml
index b774e8766ec1b..dffa71bd7c06b 100644
--- a/pulsar-io/debezium/mysql/pom.xml
+++ b/pulsar-io/debezium/mysql/pom.xml
@@ -24,7 +24,7 @@
org.apache.pulsar
pulsar-io-debezium
- 2.4.1-inst1
+ 2.4.1-inst3
pulsar-io-debezium-mysql
diff --git a/pulsar-io/debezium/pom.xml b/pulsar-io/debezium/pom.xml
index 913e4ac854496..a0af452b13004 100644
--- a/pulsar-io/debezium/pom.xml
+++ b/pulsar-io/debezium/pom.xml
@@ -25,7 +25,7 @@
org.apache.pulsar
pulsar-io
- 2.4.1-inst1
+ 2.4.1-inst3
pulsar-io-debezium
diff --git a/pulsar-io/debezium/postgres/pom.xml b/pulsar-io/debezium/postgres/pom.xml
index 6244e2176e6ef..556c4e0de97c0 100644
--- a/pulsar-io/debezium/postgres/pom.xml
+++ b/pulsar-io/debezium/postgres/pom.xml
@@ -24,7 +24,7 @@
org.apache.pulsar
pulsar-io-debezium
- 2.4.1-inst1
+ 2.4.1-inst3
pulsar-io-debezium-postgres
diff --git a/pulsar-io/docs/pom.xml b/pulsar-io/docs/pom.xml
index 26ee800b033c3..9c1da04120d4b 100644
--- a/pulsar-io/docs/pom.xml
+++ b/pulsar-io/docs/pom.xml
@@ -24,7 +24,7 @@
org.apache.pulsar
pulsar-io
- 2.4.1-inst1
+ 2.4.1-inst3
pulsar-io-docs
diff --git a/pulsar-io/elastic-search/pom.xml b/pulsar-io/elastic-search/pom.xml
index 3988408b38f33..5459813921625 100644
--- a/pulsar-io/elastic-search/pom.xml
+++ b/pulsar-io/elastic-search/pom.xml
@@ -23,7 +23,7 @@
org.apache.pulsar
pulsar-io
- 2.4.1-inst1
+ 2.4.1-inst3
pulsar-io-elastic-search
Pulsar IO :: ElasticSearch
diff --git a/pulsar-io/file/pom.xml b/pulsar-io/file/pom.xml
index 1eae7842b7b3c..28f0a3e175030 100644
--- a/pulsar-io/file/pom.xml
+++ b/pulsar-io/file/pom.xml
@@ -23,7 +23,7 @@
org.apache.pulsar
pulsar-io
- 2.4.1-inst1
+ 2.4.1-inst3
pulsar-io-file
diff --git a/pulsar-io/flume/pom.xml b/pulsar-io/flume/pom.xml
index 2f689a5a7d346..38614a19037c6 100644
--- a/pulsar-io/flume/pom.xml
+++ b/pulsar-io/flume/pom.xml
@@ -25,7 +25,7 @@
org.apache.pulsar
pulsar-io
- 2.4.1-inst1
+ 2.4.1-inst3
pulsar-io-flume
diff --git a/pulsar-io/hbase/pom.xml b/pulsar-io/hbase/pom.xml
index ad2d5bae6d03e..a43e36548f052 100644
--- a/pulsar-io/hbase/pom.xml
+++ b/pulsar-io/hbase/pom.xml
@@ -25,7 +25,7 @@
pulsar-io
org.apache.pulsar
- 2.4.1-inst1
+ 2.4.1-inst3
pulsar-io-hbase
Pulsar IO :: Hbase
diff --git a/pulsar-io/hdfs2/pom.xml b/pulsar-io/hdfs2/pom.xml
index 42c6cb1692e14..e4f876fea65cc 100644
--- a/pulsar-io/hdfs2/pom.xml
+++ b/pulsar-io/hdfs2/pom.xml
@@ -23,7 +23,7 @@
org.apache.pulsar
pulsar-io
- 2.4.1-inst1
+ 2.4.1-inst3
pulsar-io-hdfs2
Pulsar IO :: Hdfs2
diff --git a/pulsar-io/hdfs3/pom.xml b/pulsar-io/hdfs3/pom.xml
index 6657c5a8db378..e47331e426d12 100644
--- a/pulsar-io/hdfs3/pom.xml
+++ b/pulsar-io/hdfs3/pom.xml
@@ -23,7 +23,7 @@
org.apache.pulsar
pulsar-io
- 2.4.1-inst1
+ 2.4.1-inst3
pulsar-io-hdfs3
Pulsar IO :: Hdfs3
diff --git a/pulsar-io/influxdb/pom.xml b/pulsar-io/influxdb/pom.xml
index d8ed24d96f9da..acba54412c46c 100644
--- a/pulsar-io/influxdb/pom.xml
+++ b/pulsar-io/influxdb/pom.xml
@@ -25,7 +25,7 @@
pulsar-io
org.apache.pulsar
- 2.4.1-inst1
+ 2.4.1-inst3
pulsar-io-influxdb
diff --git a/pulsar-io/jdbc/pom.xml b/pulsar-io/jdbc/pom.xml
index 806a18c3e53b7..51427cb575622 100644
--- a/pulsar-io/jdbc/pom.xml
+++ b/pulsar-io/jdbc/pom.xml
@@ -24,7 +24,7 @@
org.apache.pulsar
pulsar-io
- 2.4.1-inst1
+ 2.4.1-inst3
pulsar-io-jdbc
diff --git a/pulsar-io/kafka-connect-adaptor/pom.xml b/pulsar-io/kafka-connect-adaptor/pom.xml
index 744329aa89b33..a4eedaa4f69c4 100644
--- a/pulsar-io/kafka-connect-adaptor/pom.xml
+++ b/pulsar-io/kafka-connect-adaptor/pom.xml
@@ -24,7 +24,7 @@
org.apache.pulsar
pulsar-io
- 2.4.1-inst1
+ 2.4.1-inst3
pulsar-io-kafka-connect-adaptor
diff --git a/pulsar-io/kafka/pom.xml b/pulsar-io/kafka/pom.xml
index ded9f7253f048..35e4688d8570c 100644
--- a/pulsar-io/kafka/pom.xml
+++ b/pulsar-io/kafka/pom.xml
@@ -24,7 +24,7 @@
org.apache.pulsar
pulsar-io
- 2.4.1-inst1
+ 2.4.1-inst3
pulsar-io-kafka
diff --git a/pulsar-io/kinesis/pom.xml b/pulsar-io/kinesis/pom.xml
index de0b04aaf668e..be902f79ae903 100644
--- a/pulsar-io/kinesis/pom.xml
+++ b/pulsar-io/kinesis/pom.xml
@@ -24,7 +24,7 @@
org.apache.pulsar
pulsar-io
- 2.4.1-inst1
+ 2.4.1-inst3
pulsar-io-kinesis
diff --git a/pulsar-io/mongo/pom.xml b/pulsar-io/mongo/pom.xml
index b1d53363b1169..70da0248c5bfd 100644
--- a/pulsar-io/mongo/pom.xml
+++ b/pulsar-io/mongo/pom.xml
@@ -26,7 +26,7 @@
org.apache.pulsar
pulsar-io
- 2.4.1-inst1
+ 2.4.1-inst3
pulsar-io-mongo
diff --git a/pulsar-io/netty/pom.xml b/pulsar-io/netty/pom.xml
index 635531be18851..923812ae1151f 100644
--- a/pulsar-io/netty/pom.xml
+++ b/pulsar-io/netty/pom.xml
@@ -25,7 +25,7 @@
org.apache.pulsar
pulsar-io
- 2.4.1-inst1
+ 2.4.1-inst3
pulsar-io-netty
diff --git a/pulsar-io/pom.xml b/pulsar-io/pom.xml
index f884fa6c17557..4938770c5ecc4 100644
--- a/pulsar-io/pom.xml
+++ b/pulsar-io/pom.xml
@@ -25,7 +25,7 @@
org.apache.pulsar
pulsar
- 2.4.1-inst1
+ 2.4.1-inst3
pulsar-io
diff --git a/pulsar-io/rabbitmq/pom.xml b/pulsar-io/rabbitmq/pom.xml
index fb9d44d0f71ce..087c4a9336531 100644
--- a/pulsar-io/rabbitmq/pom.xml
+++ b/pulsar-io/rabbitmq/pom.xml
@@ -24,7 +24,7 @@
org.apache.pulsar
pulsar-io
- 2.4.1-inst1
+ 2.4.1-inst3
pulsar-io-rabbitmq
diff --git a/pulsar-io/redis/pom.xml b/pulsar-io/redis/pom.xml
index 18f43d8050882..15c7f3a906007 100644
--- a/pulsar-io/redis/pom.xml
+++ b/pulsar-io/redis/pom.xml
@@ -25,7 +25,7 @@
pulsar-io
org.apache.pulsar
- 2.4.1-inst1
+ 2.4.1-inst3
pulsar-io-redis
diff --git a/pulsar-io/solr/pom.xml b/pulsar-io/solr/pom.xml
index 4f2b6598fa324..2455d0d52f7d4 100644
--- a/pulsar-io/solr/pom.xml
+++ b/pulsar-io/solr/pom.xml
@@ -25,7 +25,7 @@
pulsar-io
org.apache.pulsar
- 2.4.1-inst1
+ 2.4.1-inst3
diff --git a/pulsar-io/twitter/pom.xml b/pulsar-io/twitter/pom.xml
index f99e1ebd53be1..7f4767cdf3078 100644
--- a/pulsar-io/twitter/pom.xml
+++ b/pulsar-io/twitter/pom.xml
@@ -24,7 +24,7 @@
org.apache.pulsar
pulsar-io
- 2.4.1-inst1
+ 2.4.1-inst3
pulsar-io-twitter
diff --git a/pulsar-log4j2-appender/pom.xml b/pulsar-log4j2-appender/pom.xml
index 99f74ff877758..9c5989b667a30 100644
--- a/pulsar-log4j2-appender/pom.xml
+++ b/pulsar-log4j2-appender/pom.xml
@@ -24,7 +24,7 @@
org.apache.pulsar
pulsar
- 2.4.1-inst1
+ 2.4.1-inst3
..
diff --git a/pulsar-proxy/pom.xml b/pulsar-proxy/pom.xml
index 51fd555c532b3..5dbd7e2231ce8 100644
--- a/pulsar-proxy/pom.xml
+++ b/pulsar-proxy/pom.xml
@@ -24,7 +24,7 @@
org.apache.pulsar
pulsar
- 2.4.1-inst1
+ 2.4.1-inst3
pulsar-proxy
diff --git a/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/AdminProxyHandler.java b/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/AdminProxyHandler.java
index fdf687dfc9ea0..6a767d42a76df 100644
--- a/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/AdminProxyHandler.java
+++ b/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/AdminProxyHandler.java
@@ -20,9 +20,14 @@
import static org.apache.commons.lang3.StringUtils.isBlank;
+import java.io.ByteArrayOutputStream;
import java.io.IOException;
+import java.io.InputStream;
import java.net.URI;
+import java.nio.ByteBuffer;
import java.security.cert.X509Certificate;
+import java.util.Collections;
+import java.util.Iterator;
import java.util.Objects;
import java.util.concurrent.Executor;
@@ -30,6 +35,7 @@
import javax.servlet.ServletConfig;
import javax.servlet.ServletException;
import javax.servlet.http.HttpServletRequest;
+import javax.servlet.http.HttpServletResponse;
import org.apache.pulsar.broker.web.AuthenticationFilter;
import org.apache.pulsar.client.api.Authentication;
@@ -42,6 +48,7 @@
import org.eclipse.jetty.client.HttpRequest;
import org.eclipse.jetty.client.ProtocolHandlers;
import org.eclipse.jetty.client.RedirectProtocolHandler;
+import org.eclipse.jetty.client.api.ContentProvider;
import org.eclipse.jetty.client.api.Request;
import org.eclipse.jetty.http.HttpHeader;
import org.eclipse.jetty.proxy.ProxyServlet;
@@ -133,6 +140,33 @@ protected HttpClient createHttpClient() throws ServletException {
}
+ // This class allows the request body to be replayed, the default implementation
+ // does not
+ protected class ReplayableProxyContentProvider extends ProxyInputStreamContentProvider {
+ private Boolean firstIteratorCalled = false;
+ private final ByteArrayOutputStream bodyBuffer;
+ protected ReplayableProxyContentProvider(HttpServletRequest request, HttpServletResponse response, Request proxyRequest, InputStream input) {
+ super(request, response, proxyRequest, input);
+ bodyBuffer = new ByteArrayOutputStream(request.getContentLength());
+ }
+
+ @Override
+ public Iterator iterator() {
+ if (firstIteratorCalled) {
+ return Collections.singleton(ByteBuffer.wrap(bodyBuffer.toByteArray())).iterator();
+ } else {
+ firstIteratorCalled = true;
+ return super.iterator();
+ }
+ }
+
+ @Override
+ protected ByteBuffer onRead(byte[] buffer, int offset, int length) {
+ bodyBuffer.write(buffer, offset, length);
+ return super.onRead(buffer, offset, length);
+ }
+ }
+
private static class JettyHttpClient extends HttpClient {
public JettyHttpClient() {
super();
@@ -159,6 +193,12 @@ protected Request copyRequest(HttpRequest oldRequest, URI newURI) {
}
+ @Override
+ protected ContentProvider proxyRequestContent(HttpServletRequest request,
+ HttpServletResponse response, Request proxyRequest) throws IOException {
+ return new ReplayableProxyContentProvider(request, response, proxyRequest, request.getInputStream());
+ }
+
@Override
protected HttpClient newHttpClient() {
try {
diff --git a/pulsar-spark/pom.xml b/pulsar-spark/pom.xml
index 193a50c9c1da4..e6d7441305d9f 100644
--- a/pulsar-spark/pom.xml
+++ b/pulsar-spark/pom.xml
@@ -26,7 +26,7 @@
org.apache.pulsar
pulsar
- 2.4.1-inst1
+ 2.4.1-inst3
..
diff --git a/pulsar-sql/pom.xml b/pulsar-sql/pom.xml
index 4f13d2fc26bb7..9ccec4b3dc8c0 100644
--- a/pulsar-sql/pom.xml
+++ b/pulsar-sql/pom.xml
@@ -25,7 +25,7 @@
org.apache.pulsar
pulsar
- 2.4.1-inst1
+ 2.4.1-inst3
pulsar-sql
diff --git a/pulsar-sql/presto-distribution/pom.xml b/pulsar-sql/presto-distribution/pom.xml
index 9a4753fed8945..8367dd8971b5b 100644
--- a/pulsar-sql/presto-distribution/pom.xml
+++ b/pulsar-sql/presto-distribution/pom.xml
@@ -31,7 +31,7 @@
org.apache.pulsar
pulsar-presto-distribution
Pulsar SQL :: Pulsar Presto Distribution
- 2.4.1-inst1
+ 2.4.1-inst3
0.206
diff --git a/pulsar-sql/presto-pulsar-plugin/pom.xml b/pulsar-sql/presto-pulsar-plugin/pom.xml
index 7b9656f720b2d..6872e387776fd 100644
--- a/pulsar-sql/presto-pulsar-plugin/pom.xml
+++ b/pulsar-sql/presto-pulsar-plugin/pom.xml
@@ -25,7 +25,7 @@
org.apache.pulsar
pulsar-sql
- 2.4.1-inst1
+ 2.4.1-inst3
pulsar-presto-connector
diff --git a/pulsar-sql/presto-pulsar/pom.xml b/pulsar-sql/presto-pulsar/pom.xml
index 10c0d0c2dca5a..4dc3dfc9f7589 100644
--- a/pulsar-sql/presto-pulsar/pom.xml
+++ b/pulsar-sql/presto-pulsar/pom.xml
@@ -25,7 +25,7 @@
org.apache.pulsar
pulsar-sql
- 2.4.1-inst1
+ 2.4.1-inst3
pulsar-presto-connector-original
diff --git a/pulsar-storm/pom.xml b/pulsar-storm/pom.xml
index 6ff400061d92c..be88b3dae25db 100644
--- a/pulsar-storm/pom.xml
+++ b/pulsar-storm/pom.xml
@@ -26,7 +26,7 @@
org.apache.pulsar
pulsar
- 2.4.1-inst1
+ 2.4.1-inst3
..
diff --git a/pulsar-testclient/pom.xml b/pulsar-testclient/pom.xml
index b5084a3211ce9..d8f01a5ff508a 100644
--- a/pulsar-testclient/pom.xml
+++ b/pulsar-testclient/pom.xml
@@ -25,7 +25,7 @@
org.apache.pulsar
pulsar
- 2.4.1-inst1
+ 2.4.1-inst3
..
diff --git a/pulsar-transaction/buffer/pom.xml b/pulsar-transaction/buffer/pom.xml
index 867bc3f425f3c..dbdfa9e2d9450 100644
--- a/pulsar-transaction/buffer/pom.xml
+++ b/pulsar-transaction/buffer/pom.xml
@@ -27,7 +27,7 @@
org.apache.pulsar
pulsar-transaction-parent
- 2.4.1-inst1
+ 2.4.1-inst3
pulsar-transaction-buffer
diff --git a/pulsar-transaction/common/pom.xml b/pulsar-transaction/common/pom.xml
index c352b7df37d51..1fb12eba66969 100644
--- a/pulsar-transaction/common/pom.xml
+++ b/pulsar-transaction/common/pom.xml
@@ -27,7 +27,7 @@
org.apache.pulsar
pulsar-transaction-parent
- 2.4.1-inst1
+ 2.4.1-inst3
pulsar-transaction-common
diff --git a/pulsar-transaction/coordinator/pom.xml b/pulsar-transaction/coordinator/pom.xml
index e5dbc43c6a36a..542b29e2b1160 100644
--- a/pulsar-transaction/coordinator/pom.xml
+++ b/pulsar-transaction/coordinator/pom.xml
@@ -27,7 +27,7 @@
org.apache.pulsar
pulsar-transaction-parent
- 2.4.1-inst1
+ 2.4.1-inst3
pulsar-transaction-coordinator
diff --git a/pulsar-transaction/pom.xml b/pulsar-transaction/pom.xml
index 9c1f29b72c1a3..6893ac5c80dac 100644
--- a/pulsar-transaction/pom.xml
+++ b/pulsar-transaction/pom.xml
@@ -25,7 +25,7 @@
org.apache.pulsar
pulsar
- 2.4.1-inst1
+ 2.4.1-inst3
pulsar-transaction-parent
diff --git a/pulsar-websocket/pom.xml b/pulsar-websocket/pom.xml
index 32961884642f8..a3d209345cb56 100644
--- a/pulsar-websocket/pom.xml
+++ b/pulsar-websocket/pom.xml
@@ -25,7 +25,7 @@
org.apache.pulsar
pulsar
- 2.4.1-inst1
+ 2.4.1-inst3
..
diff --git a/pulsar-zookeeper-utils/pom.xml b/pulsar-zookeeper-utils/pom.xml
index e0ebf189b818e..c3d978600bc82 100644
--- a/pulsar-zookeeper-utils/pom.xml
+++ b/pulsar-zookeeper-utils/pom.xml
@@ -25,7 +25,7 @@
org.apache.pulsar
pulsar
- 2.4.1-inst1
+ 2.4.1-inst3
..
diff --git a/pulsar-zookeeper/pom.xml b/pulsar-zookeeper/pom.xml
index 69fce2ac7e63c..7a05de6455025 100644
--- a/pulsar-zookeeper/pom.xml
+++ b/pulsar-zookeeper/pom.xml
@@ -25,7 +25,7 @@
org.apache.pulsar
pulsar
- 2.4.1-inst1
+ 2.4.1-inst3
..
diff --git a/site2/docs/functions-cli.md b/site2/docs/functions-cli.md
new file mode 100644
index 0000000000000..4963890719338
--- /dev/null
+++ b/site2/docs/functions-cli.md
@@ -0,0 +1,197 @@
+---
+id: functions-cli
+title: Pulsar Functions command line tool
+sidebar_label: Reference: CLI
+---
+
+The following tables list Pulsar Functions command-line tools. You can learn Pulsar Functions modes, commands, and parameters.
+
+## localrun
+
+Run Pulsar Functions locally, rather than deploying it to the Pulsar cluster.
+
+Name | Description | Default
+---|---|---
+auto-ack | Whether or not the framework acknowledges messages automatically. | false |
+broker-service-url | The URL for the Pulsar broker. | |
+classname | The class name of a Pulsar Function.| |
+client-auth-params | Client authentication parameter. | |
+client-auth-plugin | Client authentication plugin using which function-process can connect to broker. | |
+CPU | The CPU in cores that need to be allocated per function instance (applicable only to docker runtime).| |
+custom-schema-inputs | The map of input topics to Schema class names (as a JSON string). | |
+custom-serde-inputs | The map of input topics to SerDe class names (as a JSON string). | |
+dead-letter-topic | The topic where all messages that were not processed successfully are sent. | |
+disk | The disk in bytes that need to be allocated per function instance (applicable only to docker runtime). | |
+fqfn | The Fully Qualified Function Name (FQFN) for the function. | |
+function-config-file | The path to a YAML config file specifying the configuration of a Pulsar Function. | |
+go | Path to the main Go executable binary for the function (if the function is written in Go). | |
+hostname-verification-enabled | Enable hostname verification. | false
+inputs | The input topic or topics of a Pulsar Function (multiple topics can be specified as a comma-separated list). | |
+jar | Path to the jar file for the function (if the function is written in Java). It also supports URL-path [http/https/file (file protocol assumes that file already exists on worker host)] from which worker can download the package. | |
+instance-id-offset | Start the instanceIds from this offset. | 0
+log-topic | The topic to which the logs a Pulsar Function are produced. | |
+max-message-retries | How many times should we try to process a message before giving up. | |
+name | The name of a Pulsar Function. | |
+namespace | The namespace of a Pulsar Function. | |
+output | The output topic of a Pulsar Function (If none is specified, no output is written). | |
+output-serde-classname | The SerDe class to be used for messages output by the function. | |
+parallelism | The parallelism factor of a Pulsar Function (i.e. the number of function instances to run). | |
+processing-guarantees | The processing guarantees (delivery semantics) applied to the function. Available values: [ATLEAST_ONCE, ATMOST_ONCE, EFFECTIVELY_ONCE]. | ATLEAST_ONCE
+py | Path to the main Python file/Python Wheel file for the function (if the function is written in Python). | |
+ram | The ram in bytes that need to be allocated per function instance (applicable only to process/docker runtime). | |
+retain-ordering | Function consumes and processes messages in order. | |
+schema-type | The builtin schema type or custom schema class name to be used for messages output by the function. |
+sliding-interval-count | The number of messages after which the window slides. | |
+sliding-interval-duration-ms | The time duration after which the window slides. | |
+subs-name | Pulsar source subscription name if user wants a specific subscription-name for the input-topic consumer. | |
+tenant | The tenant of a Pulsar Function. | |
+timeout-ms | The message timeout in milliseconds. | |
+tls-allow-insecure | Allow insecure tls connection. | false
+tls-trust-cert-path | tls trust cert file path. | |
+topics-pattern | The topic pattern to consume from list of topics under a namespace that match the pattern. [--input] and [--topic-pattern] are mutually exclusive. Add SerDe class name for a pattern in --custom-serde-inputs (only supported in Java Function). | |
+use-tls | Use tls connection. | false
+user-config | User-defined config key/values. | |
+window-length-count | The number of messages per window. | |
+window-length-duration-ms | The time duration of the window in milliseconds. | |
+
+
+## create
+
+Create and deploy a Pulsar Function in cluster mode.
+
+Name | Description | Default
+---|---|---
+auto-ack | Whether or not the framework acknowledges messages automatically. | false |
+classname | The class name of a Pulsar Function. | |
+CPU | The CPU in cores that need to be allocated per function instance (applicable only to docker runtime).| |
+custom-runtime-options | A string that encodes options to customize the runtime, see docs for configured runtime for details | |
+custom-schema-inputs | The map of input topics to Schema class names (as a JSON string). | |
+custom-serde-inputs | The map of input topics to SerDe class names (as a JSON string). | |
+dead-letter-topic | The topic where all messages that were not processed successfully are sent. | |
+disk | The disk in bytes that need to be allocated per function instance (applicable only to docker runtime). | |
+fqfn | The Fully Qualified Function Name (FQFN) for the function. | |
+function-config-file | The path to a YAML config file specifying the configuration of a Pulsar Function. | |
+go | Path to the main Go executable binary for the function (if the function is written in Go). | |
+inputs | The input topic or topics of a Pulsar Function (multiple topics can be specified as a comma-separated list). | |
+jar | Path to the jar file for the function (if the function is written in Java). It also supports URL-path [http/https/file (file protocol assumes that file already exists on worker host)] from which worker can download the package. | |
+log-topic | The topic to which the logs of a Pulsar Function are produced. | |
+max-message-retries | How many times should we try to process a message before giving up. | |
+name | The name of a Pulsar Function. | |
+namespace | The namespace of a Pulsar Function. | |
+output | The output topic of a Pulsar Function (If none is specified, no output is written). | |
+output-serde-classname | The SerDe class to be used for messages output by the function. | |
+parallelism | The parallelism factor of a Pulsar Function (i.e. the number of function instances to run). | |
+processing-guarantees | The processing guarantees (delivery semantics) applied to the function. Available values: [ATLEAST_ONCE, ATMOST_ONCE, EFFECTIVELY_ONCE]. | ATLEAST_ONCE
+py | Path to the main Python file/Python Wheel file for the function (if the function is written in Python). | |
+ram | The ram in bytes that need to be allocated per function instance (applicable only to process/docker runtime). | |
+retain-ordering | Function consumes and processes messages in order. | |
+schema-type | The builtin schema type or custom schema class name to be used for messages output by the function. |
+sliding-interval-count | The number of messages after which the window slides. | |
+sliding-interval-duration-ms | The time duration after which the window slides. | |
+subs-name | Pulsar source subscription name if user wants a specific subscription-name for the input-topic consumer. | |
+tenant | The tenant of a Pulsar Function. | |
+timeout-ms | The message timeout in milliseconds. | |
+topics-pattern | The topic pattern to consume from list of topics under a namespace that match the pattern. [--input] and [--topic-pattern] are mutually exclusive. Add SerDe class name for a pattern in --custom-serde-inputs (only supported in Java Function). | |
+user-config | User-defined config key/values. | |
+window-length-count | The number of messages per window. | |
+window-length-duration-ms | The time duration of the window in milliseconds. | |
+
+## delete
+
+Delete a Pulsar Function that is running on a Pulsar cluster.
+
+Name | Description | Default
+---|---|---
+fqfn | The Fully Qualified Function Name (FQFN) for the function. | |
+name | The name of a Pulsar Function. | |
+namespace | The namespace of a Pulsar Function. | |
+tenant | The tenant of a Pulsar Function. | |
+
+## update
+
+Update a Pulsar Function that has been deployed to a Pulsar cluster.
+
+Name | Description | Default
+---|---|---
+auto-ack | Whether or not the framework acknowledges messages automatically. | false
+classname | The class name of a Pulsar Function. | |
+CPU | The CPU in cores that need to be allocated per function instance (applicable only to docker runtime). | |
+custom-runtime-options | A string that encodes options to customize the runtime, see docs for configured runtime for details | |
+custom-schema-inputs | The map of input topics to Schema class names (as a JSON string). | |
+custom-serde-inputs | The map of input topics to SerDe class names (as a JSON string). | |
+dead-letter-topic | The topic where all messages that were not processed successfully are sent. | |
+disk | The disk in bytes that need to be allocated per function instance (applicable only to docker runtime). | |
+fqfn | The Fully Qualified Function Name (FQFN) for the function. | |
+function-config-file | The path to a YAML config file specifying the configuration of a Pulsar Function. | |
+go | Path to the main Go executable binary for the function (if the function is written in Go). | |
+inputs | The input topic or topics of a Pulsar Function (multiple topics can be specified as a comma-separated list). | |
+jar | Path to the jar file for the function (if the function is written in Java). It also supports URL-path [http/https/file (file protocol assumes that file already exists on worker host)] from which worker can download the package. | |
+log-topic | The topic to which the logs of a Pulsar Function are produced. | |
+max-message-retries | How many times should we try to process a message before giving up. | |
+name | The name of a Pulsar Function. | |
+namespace | The namespace of a Pulsar Function. | |
+output | The output topic of a Pulsar Function (If none is specified, no output is written). | |
+output-serde-classname | The SerDe class to be used for messages output by the function. | |
+parallelism | The parallelism factor of a Pulsar Function (i.e. the number of function instances to run). | |
+processing-guarantees | The processing guarantees (delivery semantics) applied to the function. Available values: [ATLEAST_ONCE, ATMOST_ONCE, EFFECTIVELY_ONCE]. | ATLEAST_ONCE
+py | Path to the main Python file/Python Wheel file for the function (if the function is written in Python). | |
+ram | The ram in bytes that need to be allocated per function instance (applicable only to process/docker runtime). | |
+retain-ordering | Function consumes and processes messages in order. | |
+schema-type | The builtin schema type or custom schema class name to be used for messages output by the function. |
+sliding-interval-count | The number of messages after which the window slides. | |
+sliding-interval-duration-ms | The time duration after which the window slides. | |
+subs-name | Pulsar source subscription name if user wants a specific subscription-name for the input-topic consumer. | |
+tenant | The tenant of a Pulsar Function. | |
+timeout-ms | The message timeout in milliseconds. | |
+topics-pattern | The topic pattern to consume from list of topics under a namespace that match the pattern. [--input] and [--topic-pattern] are mutually exclusive. Add SerDe class name for a pattern in --custom-serde-inputs (only supported in Java Function). | |
+update-auth-data | Whether or not to update the auth data. | false
+user-config | User-defined config key/values. | |
+window-length-count | The number of messages per window. | |
+window-length-duration-ms | The time duration of the window in milliseconds. | |
+
+## get
+
+Fetch information about a Pulsar Function.
+
+Name | Description | Default
+---|---|---
+fqfn | The Fully Qualified Function Name (FQFN) for the function. | |
+name | The name of a Pulsar Function. | |
+namespace | The namespace of a Pulsar Function. | |
+tenant | The tenant of a Pulsar Function. | |
+
+## restart
+
+Restart function instance.
+
+Name | Description | Default
+---|---|---
+fqfn | The Fully Qualified Function Name (FQFN) for the function. | |
+instance-id | The function instanceId (restart all instances if instance-id is not provided. | |
+name | The name of a Pulsar Function. | |
+namespace | The namespace of a Pulsar Function. | |
+tenant | The tenant of a Pulsar Function. | |
+
+## stop
+
+Stops function instance.
+
+Name | Description | Default
+---|---|---
+fqfn | The Fully Qualified Function Name (FQFN) for the function. | |
+instance-id | The function instanceId (restart all instances if instance-id is not provided. | |
+name | The name of a Pulsar Function. | |
+namespace | The namespace of a Pulsar Function. | |
+tenant | The tenant of a Pulsar Function. | |
+
+## start
+
+Starts a stopped function instance.
+
+Name | Description | Default
+---|---|---
+fqfn | The Fully Qualified Function Name (FQFN) for the function. | |
+instance-id | The function instanceId (restart all instances if instance-id is not provided. | |
+name | The name of a Pulsar Function. | |
+namespace | The namespace of a Pulsar Function. | |
+tenant | The tenant of a Pulsar Function. | |
diff --git a/site2/docs/functions-runtime.md b/site2/docs/functions-runtime.md
index 5a570264acfce..75d3aef2c9a7b 100644
--- a/site2/docs/functions-runtime.md
+++ b/site2/docs/functions-runtime.md
@@ -10,6 +10,10 @@ Currently, Pulsar Functions support the following three methods to run functions
- *Process*: Invoke functions in processes forked by Functions Worker.
- *Kubernetes*: Submit functions as Kubernetes StatefulSets by Functions Worker.
+The differences of the thread and process modes are:
+- Thread mode: when a function runs in thread mode, it runs on the same Java virtual machine (JVM) with Functions worker.
+- Process mode: when a function runs in process mode, it runs on the same machine that Functions worker runs.
+
## Configure thread runtime
*Thread* runtime is easy to configure. In most cases, you don't need to configure anything. You can customize the thread group name with the following settings:
@@ -58,7 +62,7 @@ kubernetesContainerFactory:
# setting this to true is let function worker to submit functions to the same k8s cluster as function worker
# is running. setting this to false if your function worker is not running as a k8 pod.
submittingInsidePod: false
- # setting the pulsar service url that pulsar function should use to connect to pulsar
+ # setting the pulsar service url that pulsar function should use to connect to pulsar
# if it is not set, it will use the pulsar service url configured in worker service
pulsarServiceUrl:
# setting the pulsar admin url that pulsar function should use to connect to pulsar
@@ -71,6 +75,13 @@ kubernetesContainerFactory:
extraFunctionDependenciesDir:
# Additional memory padding added on top of the memory requested by the function per on a per instance basis
percentMemoryPadding: 10
+ # an instance of `org.apache.pulsar.functions.runtime.KubernetesManifestCustomizer` which can be implemented
+ # to inject customization of the generated k8s manifest, leave blank to use the default customization plugin.
+ # This class is passed the `customRuntimeOptions` flag passed to functions
+ kubernetesManifestCustomizerClassName:
+ # a string that is passed to the manifest customizer class, this config is used for all functions and is static
+ # (as opposed to the `customRuntimeOptions` which can be provided per function)
+ kubernetesManifestCustomizerConfig:
```
If you have already run a Pulsar cluster on Kubernetes, you can keep the settings unchanged at most of time.
@@ -79,7 +90,7 @@ However, if you enable RBAC on deploying your Pulsar cluster, make sure the serv
running Functions Workers (or brokers, if Functions Workers run along with brokers) have permissions on the following
kubernetes APIs.
-- services
+- services
- configmaps
- pods
- apps.statefulsets
@@ -104,7 +115,7 @@ If this happens, you need to grant the required permissions to the service accou
apiVersion: rbac.authorization.k8s.io/v1beta1
kind: ClusterRole
metadata:
- name: functions-worker
+ name: functions-worker
rules:
- apiGroups: [""]
resources:
@@ -136,4 +147,35 @@ roleRef:
subjects:
- kind: ServiceAccount
name: functions-worker
-```
\ No newline at end of file
+```
+
+### Kubernetes CustomRuntimeOptions
+
+The functions (and sinks/sources) API provides a flag, `customRuntimeOptions` which can be used to pass options to the runtime to customize how the runtime operates.
+
+In the case of case of kubernetes, this is passed to an instance of the `org.apache.pulsar.functions.runtime.KubernetesManifestCustomizer`. This interface can be overridden
+and allows for a high degree of customization over how the K8S manifests are generated.
+
+Pulsar provides a default implementation of this interface which takes the following options as a json string:
+
+```Json
+{
+ "jobNamespace": "namespace", // the k8s namespace to run this function in
+ "extractLabels": { // extra labels to attach to the statefulSet, service, and pods
+ "extraLabel": "value"
+ },
+ "extraAnnotations": { // extra annotations to attach to the statefulSet, service, and pods
+ "extraAnnotation": "value"
+ },
+ "nodeSelectorLabels": { // node selector labels to add on to the pod spec
+ "customLabel": "value"
+ },
+ "tolerations": [ // tolerations to add to the pod spec
+ {
+ "key": "custom-key",
+ "value": "value",
+ "effect": "NoSchedule"
+ }
+ ]
+}
+```
diff --git a/tests/bc_2_0_0/pom.xml b/tests/bc_2_0_0/pom.xml
index 6af3fc0b6c595..04431c5992365 100644
--- a/tests/bc_2_0_0/pom.xml
+++ b/tests/bc_2_0_0/pom.xml
@@ -25,7 +25,7 @@
org.apache.pulsar.tests
tests-parent
- 2.4.1-inst1
+ 2.4.1-inst3
bc_2_0_0
diff --git a/tests/bc_2_0_1/pom.xml b/tests/bc_2_0_1/pom.xml
index 774d0a60c03ae..22351b46a8dcb 100644
--- a/tests/bc_2_0_1/pom.xml
+++ b/tests/bc_2_0_1/pom.xml
@@ -25,7 +25,7 @@
org.apache.pulsar.tests
tests-parent
- 2.4.1-inst1
+ 2.4.1-inst3
bc_2_0_1
diff --git a/tests/docker-images/java-test-functions/pom.xml b/tests/docker-images/java-test-functions/pom.xml
index 227ec58e81cca..0f753822b5fa3 100644
--- a/tests/docker-images/java-test-functions/pom.xml
+++ b/tests/docker-images/java-test-functions/pom.xml
@@ -23,7 +23,7 @@
org.apache.pulsar.tests
docker-images
- 2.4.1-inst1
+ 2.4.1-inst3
4.0.0
org.apache.pulsar.tests
diff --git a/tests/docker-images/latest-version-image/pom.xml b/tests/docker-images/latest-version-image/pom.xml
index 148bedc668d90..e0d28b3658296 100644
--- a/tests/docker-images/latest-version-image/pom.xml
+++ b/tests/docker-images/latest-version-image/pom.xml
@@ -23,7 +23,7 @@
org.apache.pulsar.tests
docker-images
- 2.4.1-inst1
+ 2.4.1-inst3
4.0.0
org.apache.pulsar.tests
diff --git a/tests/docker-images/pom.xml b/tests/docker-images/pom.xml
index 066f0157b13dc..9badd11fcdc7e 100644
--- a/tests/docker-images/pom.xml
+++ b/tests/docker-images/pom.xml
@@ -26,7 +26,7 @@
org.apache.pulsar.tests
tests-parent
- 2.4.1-inst1
+ 2.4.1-inst3
org.apache.pulsar.tests
docker-images
diff --git a/tests/integration/pom.xml b/tests/integration/pom.xml
index 8f4c6b34d5e8a..f45d97aa5c2b7 100644
--- a/tests/integration/pom.xml
+++ b/tests/integration/pom.xml
@@ -25,7 +25,7 @@
org.apache.pulsar.tests
tests-parent
- 2.4.1-inst1
+ 2.4.1-inst3
integration
diff --git a/tests/pom.xml b/tests/pom.xml
index ecd17265bc4f5..9e0057cf015fa 100644
--- a/tests/pom.xml
+++ b/tests/pom.xml
@@ -26,7 +26,7 @@
org.apache.pulsar
pulsar
- 2.4.1-inst1
+ 2.4.1-inst3
org.apache.pulsar.tests
tests-parent
diff --git a/tests/pulsar-kafka-compat-client-test/pom.xml b/tests/pulsar-kafka-compat-client-test/pom.xml
index 34575a6caa044..d0e5bf0efb2bb 100644
--- a/tests/pulsar-kafka-compat-client-test/pom.xml
+++ b/tests/pulsar-kafka-compat-client-test/pom.xml
@@ -25,7 +25,7 @@
org.apache.pulsar.tests
tests-parent
- 2.4.1-inst1
+ 2.4.1-inst3
pulsar-kafka-compat-client-test
diff --git a/tests/pulsar-spark-test/pom.xml b/tests/pulsar-spark-test/pom.xml
index 274978dcc18fd..4db5ea640f359 100644
--- a/tests/pulsar-spark-test/pom.xml
+++ b/tests/pulsar-spark-test/pom.xml
@@ -26,7 +26,7 @@
org.apache.pulsar.tests
tests-parent
- 2.4.1-inst1
+ 2.4.1-inst3
pulsar-spark-test
diff --git a/tests/pulsar-storm-test/pom.xml b/tests/pulsar-storm-test/pom.xml
index 077c86b4fc851..ae0bdde75cfeb 100644
--- a/tests/pulsar-storm-test/pom.xml
+++ b/tests/pulsar-storm-test/pom.xml
@@ -26,7 +26,7 @@
org.apache.pulsar.tests
tests-parent
- 2.4.1-inst1
+ 2.4.1-inst3
pulsar-storm-test
diff --git a/tiered-storage/jcloud/pom.xml b/tiered-storage/jcloud/pom.xml
index 3f4bf2e561fd8..181198ccbb172 100644
--- a/tiered-storage/jcloud/pom.xml
+++ b/tiered-storage/jcloud/pom.xml
@@ -25,7 +25,7 @@
org.apache.pulsar
tiered-storage-parent
- 2.4.1-inst1
+ 2.4.1-inst3
..
diff --git a/tiered-storage/pom.xml b/tiered-storage/pom.xml
index d4a7c2b10af7a..1e154304ec796 100644
--- a/tiered-storage/pom.xml
+++ b/tiered-storage/pom.xml
@@ -25,7 +25,7 @@
org.apache.pulsar
pulsar
- 2.4.1-inst1
+ 2.4.1-inst3
..