You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by GitBox <gi...@apache.org> on 2018/06/07 05:00:03 UTC
[GitHub] rdhabalia closed pull request #1903: Support pulsar source to start
consumer with topic patterns
rdhabalia closed pull request #1903: Support pulsar source to start consumer with topic patterns
URL: https://github.com/apache/incubator-pulsar/pull/1903
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git a/pulsar-client-tools-test/src/test/java/org/apache/pulsar/admin/cli/CmdFunctionsTest.java b/pulsar-client-tools-test/src/test/java/org/apache/pulsar/admin/cli/CmdFunctionsTest.java
index 20499e4212..5861e22fa0 100644
--- a/pulsar-client-tools-test/src/test/java/org/apache/pulsar/admin/cli/CmdFunctionsTest.java
+++ b/pulsar-client-tools-test/src/test/java/org/apache/pulsar/admin/cli/CmdFunctionsTest.java
@@ -205,6 +205,31 @@ public void testCreateFunction() throws Exception {
}
+ @Test
+ public void testCreateFunctionWithTopicPatterns() throws Exception {
+ String fnName = TEST_NAME + "-function";
+ String topicPatterns = "persistent://tenant/ns/topicPattern*";
+ String outputTopicName = TEST_NAME + "-output-topic";
+ cmd.run(new String[] {
+ "create",
+ "--name", fnName,
+ "--topicsPattern", topicPatterns,
+ "--output", outputTopicName,
+ "--jar", "SomeJar.jar",
+ "--tenant", "sample",
+ "--namespace", "ns1",
+ "--className", DummyFunction.class.getName(),
+ });
+
+ CreateFunction creater = cmd.getCreater();
+ assertEquals(fnName, creater.getFunctionName());
+ assertEquals(topicPatterns, creater.getTopicsPattern());
+ assertEquals(outputTopicName, creater.getOutput());
+
+ verify(functions, times(1)).createFunction(any(FunctionDetails.class), anyString());
+
+ }
+
@Test
public void testCreateWithoutTenant() throws Exception {
String fnName = TEST_NAME + "-function";
diff --git a/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdFunctions.java b/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdFunctions.java
index a200de8ed9..69157746ce 100644
--- a/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdFunctions.java
+++ b/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdFunctions.java
@@ -18,20 +18,28 @@
*/
package org.apache.pulsar.admin.cli;
-import com.beust.jcommander.Parameter;
-import com.beust.jcommander.ParameterException;
-import com.beust.jcommander.Parameters;
-import com.beust.jcommander.converters.StringConverter;
-import com.google.common.annotations.VisibleForTesting;
-import com.google.gson.Gson;
-import com.google.gson.GsonBuilder;
-import com.google.gson.JsonParser;
-import com.google.gson.reflect.TypeToken;
-import io.netty.buffer.ByteBuf;
-import io.netty.buffer.ByteBufUtil;
-import io.netty.buffer.Unpooled;
-import lombok.Getter;
-import lombok.extern.slf4j.Slf4j;
+import static com.google.common.base.Preconditions.checkNotNull;
+import static java.nio.charset.StandardCharsets.UTF_8;
+import static java.util.Objects.isNull;
+import static org.apache.bookkeeper.common.concurrent.FutureUtils.result;
+import static org.apache.pulsar.common.naming.TopicName.DEFAULT_NAMESPACE;
+import static org.apache.pulsar.common.naming.TopicName.PUBLIC_TENANT;
+
+import java.io.File;
+import java.io.IOException;
+import java.lang.reflect.Type;
+import java.net.MalformedURLException;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.Timer;
+import java.util.TimerTask;
+import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
+
import org.apache.bookkeeper.api.StorageClient;
import org.apache.bookkeeper.api.kv.Table;
import org.apache.bookkeeper.api.kv.result.KeyValue;
@@ -59,28 +67,22 @@
import org.apache.pulsar.functions.windowing.WindowFunctionExecutor;
import org.apache.pulsar.functions.windowing.WindowUtils;
-import java.io.File;
-import java.io.IOException;
-import java.lang.reflect.Type;
-import java.net.MalformedURLException;
-import java.util.Arrays;
-import java.util.HashMap;
-import java.util.LinkedList;
-import java.util.List;
-import java.util.Map;
-import java.util.Timer;
-import java.util.TimerTask;
-import java.util.UUID;
-import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.TimeUnit;
+import com.beust.jcommander.Parameter;
+import com.beust.jcommander.ParameterException;
+import com.beust.jcommander.Parameters;
+import com.beust.jcommander.converters.StringConverter;
+import com.google.common.annotations.VisibleForTesting;
+import com.google.gson.Gson;
+import com.google.gson.GsonBuilder;
+import com.google.gson.JsonParser;
+import com.google.gson.reflect.TypeToken;
-import static com.google.common.base.Preconditions.checkNotNull;
-import static java.nio.charset.StandardCharsets.UTF_8;
-import static java.util.Objects.isNull;
-import static org.apache.bookkeeper.common.concurrent.FutureUtils.result;
-import static org.apache.pulsar.common.naming.TopicName.DEFAULT_NAMESPACE;
-import static org.apache.pulsar.common.naming.TopicName.PUBLIC_TENANT;
import static org.apache.pulsar.functions.utils.Utils.fileExists;
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.ByteBufUtil;
+import io.netty.buffer.Unpooled;
+import lombok.Getter;
+import lombok.extern.slf4j.Slf4j;
@Slf4j
@Parameters(commandDescription = "Interface for managing Pulsar Functions (lightweight, Lambda-style compute processes that work with Pulsar)")
@@ -217,6 +219,8 @@ void processArguments() throws Exception {
protected String pyFile;
@Parameter(names = "--inputs", description = "The function's input topic or topics (multiple topics can be specified as a comma-separated list)")
protected String inputs;
+ @Parameter(names = "--topicsPattern", description = "TopicsPattern to consume from list of topics under a namespace that match the pattern. [--input] and [--topicsPattern] are mutually exclusive. Add SerDe class name for a pattern in --customSerdeInputs (supported for java fun only)")
+ protected String topicsPattern;
@Parameter(names = "--output", description = "The function's output topic")
protected String output;
@Parameter(names = "--logTopic", description = "The topic to which the function's logs are produced")
@@ -288,6 +292,9 @@ void processArguments() throws Exception {
Map<String, String> customSerdeInputMap = new Gson().fromJson(customSerdeInputString, type);
functionConfig.setCustomSerdeInputs(customSerdeInputMap);
}
+ if (null != topicsPattern) {
+ functionConfig.setTopicsPattern(topicsPattern);
+ }
if (null != output) {
functionConfig.setOutput(output);
}
@@ -517,6 +524,9 @@ protected FunctionDetails convert(FunctionConfig functionConfig)
topicToSerDeClassNameMap.putAll(functionConfig.getCustomSerdeInputs());
functionConfig.getInputs().forEach(v -> topicToSerDeClassNameMap.put(v, ""));
sourceSpecBuilder.putAllTopicsToSerDeClassName(topicToSerDeClassNameMap);
+ if (StringUtils.isNotBlank(functionConfig.getTopicsPattern())) {
+ sourceSpecBuilder.setTopicsPattern(functionConfig.getTopicsPattern());
+ }
// Set subscription type based on processing semantics
if (functionConfig.getProcessingGuarantees() != null) {
diff --git a/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdSinks.java b/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdSinks.java
index 14a6fb107f..47f4ed7e4f 100644
--- a/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdSinks.java
+++ b/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdSinks.java
@@ -137,6 +137,8 @@ void runCmd() throws Exception {
protected String className;
@Parameter(names = "--inputs", description = "The sink's input topic or topics (multiple topics can be specified as a comma-separated list)")
protected String inputs;
+ @Parameter(names = "--topicsPattern", description = "TopicsPattern to consume from list of topics under a namespace that match the pattern. [--input] and [--topicsPattern] are mutually exclusive. Add SerDe class name for a pattern in --customSerdeInputs (supported for java fun only)")
+ protected String topicsPattern;
@Parameter(names = "--customSerdeInputs", description = "The map of input topics to SerDe class names (as a JSON string)")
protected String customSerdeInputString;
@Parameter(names = "--processingGuarantees", description = "The processing guarantees (aka delivery semantics) applied to the Sink")
@@ -191,6 +193,7 @@ void processArguments() throws Exception {
if (null != processingGuarantees) {
sinkConfig.setProcessingGuarantees(processingGuarantees);
}
+
Map<String, String> topicsToSerDeClassName = new HashMap<>();
if (null != inputs) {
List<String> inputTopics = Arrays.asList(inputs.split(","));
@@ -204,6 +207,10 @@ void processArguments() throws Exception {
});
}
sinkConfig.setTopicToSerdeClassName(topicsToSerDeClassName);
+
+ if (null != topicsPattern) {
+ sinkConfig.setTopicsPattern(topicsPattern);
+ }
if (parallelism != null) {
sinkConfig.setParallelism(parallelism);
@@ -299,6 +306,7 @@ protected FunctionDetails createSinkConfig(SinkConfig sinkConfig) {
SourceSpec.Builder sourceSpecBuilder = SourceSpec.newBuilder();
sourceSpecBuilder.setSubscriptionType(Function.SubscriptionType.SHARED);
sourceSpecBuilder.putAllTopicsToSerDeClassName(sinkConfig.getTopicToSerdeClassName());
+ sourceSpecBuilder.setTopicsPattern(sinkConfig.getTopicsPattern());
sourceSpecBuilder.setTypeClassName(typeArg.getName());
functionDetailsBuilder.setAutoAck(true);
functionDetailsBuilder.setSource(sourceSpecBuilder);
diff --git a/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/JavaInstanceRunnable.java b/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/JavaInstanceRunnable.java
index 6bb689c2b3..b79d508aa1 100644
--- a/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/JavaInstanceRunnable.java
+++ b/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/JavaInstanceRunnable.java
@@ -465,6 +465,7 @@ public void setupInput() throws Exception {
if (sourceSpec.getClassName().isEmpty()) {
PulsarSourceConfig pulsarSourceConfig = new PulsarSourceConfig();
pulsarSourceConfig.setTopicSerdeClassNameMap(sourceSpec.getTopicsToSerDeClassNameMap());
+ pulsarSourceConfig.setTopicsPattern(sourceSpec.getTopicsPattern());
pulsarSourceConfig.setSubscriptionName(
FunctionDetailsUtils.getFullyQualifiedName(this.instanceConfig.getFunctionDetails()));
pulsarSourceConfig.setProcessingGuarantees(
diff --git a/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/source/PulsarSource.java b/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/source/PulsarSource.java
index adc28a1330..74f23663ef 100644
--- a/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/source/PulsarSource.java
+++ b/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/source/PulsarSource.java
@@ -22,6 +22,8 @@
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;
import net.jodah.typetools.TypeResolver;
+
+import static org.apache.commons.lang3.StringUtils.isNotBlank;
import org.apache.pulsar.client.api.ConsumerBuilder;
import org.apache.pulsar.client.api.PulsarClient;
import org.apache.pulsar.client.impl.MessageIdImpl;
@@ -47,6 +49,7 @@
private PulsarClient pulsarClient;
private PulsarSourceConfig pulsarSourceConfig;
private Map<String, SerDe> topicToSerDeMap = new HashMap<>();
+ private boolean isTopicsPattern;
@Getter
private org.apache.pulsar.client.api.Consumer inputConsumer;
@@ -63,10 +66,16 @@ public void open(Map<String, Object> config) throws Exception {
// Setup pulsar consumer
ConsumerBuilder<byte[]> consumerBuilder = this.pulsarClient.newConsumer()
- .topics(new ArrayList<>(this.pulsarSourceConfig.getTopicSerdeClassNameMap().keySet()))
.subscriptionName(this.pulsarSourceConfig.getSubscriptionName())
.subscriptionType(this.pulsarSourceConfig.getSubscriptionType());
+ if(isNotBlank(this.pulsarSourceConfig.getTopicsPattern())) {
+ consumerBuilder.topicsPattern(this.pulsarSourceConfig.getTopicsPattern());
+ isTopicsPattern = true;
+ }else {
+ consumerBuilder.topics(new ArrayList<>(this.pulsarSourceConfig.getTopicSerdeClassNameMap().keySet()));
+ }
+
if (pulsarSourceConfig.getTimeoutMs() != null) {
consumerBuilder.ackTimeout(pulsarSourceConfig.getTimeoutMs(), TimeUnit.MILLISECONDS);
}
@@ -94,9 +103,19 @@ public void open(Map<String, Object> config) throws Exception {
Object object;
try {
- object = this.topicToSerDeMap.get(topicName).deserialize(message.getData());
+ SerDe deserializer = null;
+ if (this.topicToSerDeMap.containsKey(topicName)) {
+ deserializer = this.topicToSerDeMap.get(topicName);
+ } else if (isTopicsPattern) {
+ deserializer = this.topicToSerDeMap.get(this.pulsarSourceConfig.getTopicsPattern());
+ }
+ if (deserializer != null) {
+ object = deserializer.deserialize(message.getData());
+ } else {
+ throw new IllegalStateException("Topic deserializer not configured : " + topicName);
+ }
} catch (Exception e) {
- //TODO Add deserialization exception stats
+ // TODO Add deserialization exception stats
throw new RuntimeException("Error occured when attempting to deserialize input:", e);
}
diff --git a/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/source/PulsarSourceConfig.java b/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/source/PulsarSourceConfig.java
index 6c45486603..8f815f1c5b 100644
--- a/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/source/PulsarSourceConfig.java
+++ b/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/source/PulsarSourceConfig.java
@@ -38,6 +38,7 @@
SubscriptionType subscriptionType;
private String subscriptionName;
private Map<String, String> topicSerdeClassNameMap;
+ private String topicsPattern;
private String typeClassName;
private Long timeoutMs;
diff --git a/pulsar-functions/instance/src/main/python/python_instance_main.py b/pulsar-functions/instance/src/main/python/python_instance_main.py
index ca490f8d93..b1ca25fd7d 100644
--- a/pulsar-functions/instance/src/main/python/python_instance_main.py
+++ b/pulsar-functions/instance/src/main/python/python_instance_main.py
@@ -72,6 +72,7 @@ def main():
parser.add_argument('--log_topic', required=False, help='Topic to send Log Messages')
parser.add_argument('--source_subscription_type', required=True, help='Subscription Type')
parser.add_argument('--source_topics_serde_classname', required=True, help='A mapping of Input topics to SerDe')
+ parser.add_argument('--topics_pattern', required=False, help='TopicsPattern to consume from list of topics under a namespace that match the pattern (not supported)')
parser.add_argument('--source_timeout_ms', required=False, help='Source message timeout in milliseconds')
parser.add_argument('--sink_topic', required=False, help='Sink Topic')
parser.add_argument('--sink_serde_classname', required=False, help='Sink SerDe classname')
@@ -91,6 +92,8 @@ def main():
function_details.name = args.name
function_details.className = args.function_classname
+ if args.topics_pattern:
+ raise ValueError('topics_pattern is not supported by python client')
sourceSpec = Function_pb2.SourceSpec()
sourceSpec.subscriptionType = Function_pb2.SubscriptionType.Value(args.source_subscription_type)
try:
diff --git a/pulsar-functions/instance/src/test/java/org/apache/pulsar/functions/source/PulsarSourceTest.java b/pulsar-functions/instance/src/test/java/org/apache/pulsar/functions/source/PulsarSourceTest.java
index 4c81016aa7..3c5e61b0de 100644
--- a/pulsar-functions/instance/src/test/java/org/apache/pulsar/functions/source/PulsarSourceTest.java
+++ b/pulsar-functions/instance/src/test/java/org/apache/pulsar/functions/source/PulsarSourceTest.java
@@ -159,7 +159,7 @@ public void testInconsistentInputType() throws IOException {
* Verify that Default Serializer works fine.
*/
@Test
- public void testDefaultSerDe() throws PulsarClientException {
+ public void testDefaultSerDe() throws Exception {
PulsarSourceConfig pulsarConfig = getPulsarConfigs();
// set type to void
@@ -168,20 +168,14 @@ public void testDefaultSerDe() throws PulsarClientException {
pulsarConfig.setTopicSerdeClassNameMap(topicSerdeClassNameMap);
PulsarSource pulsarSource = new PulsarSource(getPulsarClient(), pulsarConfig);
- try {
- pulsarSource.open(new HashMap<>());
- } catch (Exception ex) {
- ex.printStackTrace();
- assertEquals(ex, null);
- assertTrue(false);
- }
+ pulsarSource.open(new HashMap<>());
}
/**
* Verify that Explicit setting of Default Serializer works fine.
*/
@Test
- public void testExplicitDefaultSerDe() throws PulsarClientException {
+ public void testExplicitDefaultSerDe() throws Exception {
PulsarSourceConfig pulsarConfig = getPulsarConfigs();
// set type to void
pulsarConfig.setTypeClassName(String.class.getName());
@@ -189,13 +183,7 @@ public void testExplicitDefaultSerDe() throws PulsarClientException {
pulsarConfig.setTopicSerdeClassNameMap(topicSerdeClassNameMap);
PulsarSource pulsarSource = new PulsarSource(getPulsarClient(), pulsarConfig);
- try {
- pulsarSource.open(new HashMap<>());
- } catch (Exception ex) {
- ex.printStackTrace();
- assertEquals(ex, null);
- assertTrue(false);
- }
+ pulsarSource.open(new HashMap<>());
}
@Test
@@ -210,7 +198,6 @@ public void testComplexOuputType() throws PulsarClientException {
try {
pulsarSource.setupSerDe();
} catch (Exception ex) {
- ex.printStackTrace();
fail();
}
}
diff --git a/pulsar-functions/proto/src/main/proto/Function.proto b/pulsar-functions/proto/src/main/proto/Function.proto
index f9865c0d88..0f8b649d65 100644
--- a/pulsar-functions/proto/src/main/proto/Function.proto
+++ b/pulsar-functions/proto/src/main/proto/Function.proto
@@ -70,6 +70,7 @@ message SourceSpec {
SubscriptionType subscriptionType = 3;
map<string,string> topicsToSerDeClassName = 4;
uint64 timeoutMs = 6;
+ string topicsPattern = 7;
}
message SinkSpec {
diff --git a/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/JavaInstanceMain.java b/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/JavaInstanceMain.java
index 9a17567842..b602a1d263 100644
--- a/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/JavaInstanceMain.java
+++ b/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/JavaInstanceMain.java
@@ -28,6 +28,8 @@
import io.grpc.ServerBuilder;
import io.grpc.stub.StreamObserver;
import lombok.extern.slf4j.Slf4j;
+
+import static org.apache.commons.lang3.StringUtils.isNotBlank;
import org.apache.pulsar.functions.instance.InstanceConfig;
import org.apache.pulsar.functions.proto.Function;
import org.apache.pulsar.functions.proto.Function.ProcessingGuarantees;
@@ -110,6 +112,9 @@
@Parameter(names = "--source_topics_serde_classname", description = "A map of topics to SerDe for the source")
protected String sourceTopicsSerdeClassName;
+
+ @Parameter(names = "--topics_pattern", description = "TopicsPattern to consume from list of topics under a namespace that match the pattern. [--input] and [--topicsPattern] are mutually exclusive. Add SerDe class name for a pattern in --customSerdeInputs")
+ protected String topicsPattern;
@Parameter(names = "--source_timeout_ms", description = "Source message timeout in milliseconds")
protected Long sourceTimeoutMs;
@@ -172,6 +177,9 @@ public void start() throws Exception {
}
sourceDetailsBuilder.setSubscriptionType(Function.SubscriptionType.valueOf(sourceSubscriptionType));
sourceDetailsBuilder.putAllTopicsToSerDeClassName(new Gson().fromJson(sourceTopicsSerdeClassName, Map.class));
+ if (isNotBlank(topicsPattern)) {
+ sourceDetailsBuilder.setTopicsPattern(topicsPattern);
+ }
sourceDetailsBuilder.setTypeClassName(sourceTypeClassName);
if (sourceTimeoutMs != null) {
sourceDetailsBuilder.setTimeoutMs(sourceTimeoutMs);
diff --git a/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/ProcessRuntime.java b/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/ProcessRuntime.java
index 82b2d3b33f..715eab62a7 100644
--- a/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/ProcessRuntime.java
+++ b/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/ProcessRuntime.java
@@ -28,6 +28,8 @@
import io.grpc.ManagedChannelBuilder;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;
+
+import static org.apache.commons.lang3.StringUtils.isNotBlank;
import org.apache.pulsar.functions.instance.InstanceConfig;
import org.apache.pulsar.functions.proto.Function;
import org.apache.pulsar.functions.proto.InstanceCommunication;
@@ -169,6 +171,10 @@
args.add(instanceConfig.getFunctionDetails().getSource().getSubscriptionType().toString());
args.add("--source_topics_serde_classname");
args.add(new Gson().toJson(instanceConfig.getFunctionDetails().getSource().getTopicsToSerDeClassNameMap()));
+ if (isNotBlank(instanceConfig.getFunctionDetails().getSource().getTopicsPattern())) {
+ args.add("--topics_pattern");
+ args.add(instanceConfig.getFunctionDetails().getSource().getTopicsPattern());
+ }
// sink related configs
if (instanceConfig.getFunctionDetails().getRuntime() == Function.FunctionDetails.Runtime.JAVA) {
diff --git a/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/RuntimeSpawner.java b/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/RuntimeSpawner.java
index 60b41def06..8b60deda25 100644
--- a/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/RuntimeSpawner.java
+++ b/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/RuntimeSpawner.java
@@ -31,10 +31,13 @@
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;
+
+import org.apache.commons.lang3.StringUtils;
import org.apache.pulsar.functions.instance.InstanceConfig;
import org.apache.pulsar.functions.proto.InstanceCommunication;
import org.apache.pulsar.functions.proto.InstanceCommunication.FunctionStatus;
import org.apache.pulsar.functions.utils.Utils;
+import static org.apache.pulsar.functions.proto.Function.FunctionDetails.Runtime.PYTHON;
@Slf4j
public class RuntimeSpawner implements AutoCloseable {
@@ -64,6 +67,13 @@ public RuntimeSpawner(InstanceConfig instanceConfig,
public void start() throws Exception {
log.info("RuntimeSpawner starting function {} - {}", this.instanceConfig.getFunctionDetails().getName(),
this.instanceConfig.getInstanceId());
+
+ if (instanceConfig.getFunctionDetails().getRuntime() == PYTHON
+ && instanceConfig.getFunctionDetails().getSource() != null
+ && StringUtils.isNotBlank(instanceConfig.getFunctionDetails().getSource().getTopicsPattern())) {
+ throw new IllegalArgumentException("topics-pattern is not supported for python function");
+ }
+
runtime = runtimeFactory.createContainer(this.instanceConfig, codeFile);
runtime.start();
diff --git a/pulsar-functions/runtime/src/test/java/org/apache/pulsar/functions/runtime/ProcessRuntimeTest.java b/pulsar-functions/runtime/src/test/java/org/apache/pulsar/functions/runtime/ProcessRuntimeTest.java
index 5a3732889d..65e8d89368 100644
--- a/pulsar-functions/runtime/src/test/java/org/apache/pulsar/functions/runtime/ProcessRuntimeTest.java
+++ b/pulsar-functions/runtime/src/test/java/org/apache/pulsar/functions/runtime/ProcessRuntimeTest.java
@@ -91,6 +91,7 @@ FunctionDetails createFunctionDetails(FunctionDetails.Runtime runtime) {
functionDetailsBuilder.setSource(Function.SourceSpec.newBuilder()
.setSubscriptionType(Function.SubscriptionType.FAILOVER)
.putAllTopicsToSerDeClassName(topicsToSerDeClassName)
+ .setTopicsPattern("persistent://tenant/ns/.*")
.setClassName("org.pulsar.pulsar.TestSource")
.setTypeClassName(String.class.getName()));
return functionDetailsBuilder.build();
@@ -114,7 +115,7 @@ public void testJavaConstructor() {
ProcessRuntime container = factory.createContainer(config, userJarFile);
List<String> args = container.getProcessArgs();
- assertEquals(args.size(), 51);
+ assertEquals(args.size(), 53);
String expectedArgs = "java -cp " + javaInstanceJarFile + " -Dlog4j.configurationFile=java_instance_log4j2.yml "
+ "-Dpulsar.log.dir=" + logDirectory + "/functions" + " -Dpulsar.log.file=" + config.getFunctionDetails().getName()
+ " org.apache.pulsar.functions.runtime.JavaInstanceMain"
@@ -133,6 +134,7 @@ public void testJavaConstructor() {
+ " --source_type_classname " + config.getFunctionDetails().getSource().getTypeClassName()
+ " --source_subscription_type " + config.getFunctionDetails().getSource().getSubscriptionType().name()
+ " --source_topics_serde_classname " + new Gson().toJson(topicsToSerDeClassName)
+ + " --topics_pattern " + config.getFunctionDetails().getSource().getTopicsPattern()
+ " --sink_classname " + config.getFunctionDetails().getSink().getClassName()
+ " --sink_type_classname " + config.getFunctionDetails().getSink().getTypeClassName()
+ " --sink_topic " + config.getFunctionDetails().getSink().getTopic()
@@ -146,7 +148,7 @@ public void testPythonConstructor() {
ProcessRuntime container = factory.createContainer(config, userJarFile);
List<String> args = container.getProcessArgs();
- assertEquals(args.size(), 42);
+ assertEquals(args.size(), 44);
String expectedArgs = "python " + pythonInstanceFile
+ " --py " + userJarFile + " --logging_directory "
+ logDirectory + "/functions" + " --logging_file " + config.getFunctionDetails().getName() + " --instance_id "
@@ -162,6 +164,7 @@ public void testPythonConstructor() {
+ " --max_buffered_tuples 1024 --port " + args.get(33)
+ " --source_subscription_type " + config.getFunctionDetails().getSource().getSubscriptionType().name()
+ " --source_topics_serde_classname " + new Gson().toJson(topicsToSerDeClassName)
+ + " --topics_pattern " + config.getFunctionDetails().getSource().getTopicsPattern()
+ " --sink_topic " + config.getFunctionDetails().getSink().getTopic()
+ " --sink_serde_classname " + config.getFunctionDetails().getSink().getSerDeClassName();
assertEquals(expectedArgs, String.join(" ", args));
diff --git a/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/FunctionConfig.java b/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/FunctionConfig.java
index 9671d50cf9..3140f9535c 100644
--- a/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/FunctionConfig.java
+++ b/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/FunctionConfig.java
@@ -78,6 +78,8 @@
@isMapEntryCustom(keyValidatorClasses = { ValidatorImpls.TopicNameValidator.class }, targetRuntime = ConfigValidation.Runtime.PYTHON)
private Map<String, String> customSerdeInputs;
@isValidTopicName
+ private String topicsPattern;
+ @isValidTopicName
private String output;
@isImplementationOfClass(implementsClass = SerDe.class)
private String outputSerdeClassName;
diff --git a/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/SinkConfig.java b/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/SinkConfig.java
index f4ee77cce4..097d62694c 100644
--- a/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/SinkConfig.java
+++ b/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/SinkConfig.java
@@ -30,6 +30,7 @@
import org.apache.pulsar.functions.utils.validation.ConfigValidationAnnotations.isPositiveNumber;
import org.apache.pulsar.functions.utils.validation.ConfigValidationAnnotations.isValidResources;
import org.apache.pulsar.functions.utils.validation.ConfigValidationAnnotations.isValidSinkConfig;
+import org.apache.pulsar.functions.utils.validation.ConfigValidationAnnotations.isValidTopicName;
import org.apache.pulsar.functions.utils.validation.ValidatorImpls;
import org.apache.pulsar.io.core.Sink;
@@ -55,6 +56,8 @@
@isMapEntryCustom(keyValidatorClasses = { ValidatorImpls.TopicNameValidator.class },
valueValidatorClasses = { ValidatorImpls.SerdeValidator.class })
private Map<String, String> topicToSerdeClassName;
+ @isValidTopicName
+ private String topicsPattern;
private Map<String, Object> configs = new HashMap<>();
@isPositiveNumber
private int parallelism = 1;
diff --git a/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/Utils.java b/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/Utils.java
index 2601ca0aeb..c5eba1f802 100644
--- a/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/Utils.java
+++ b/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/Utils.java
@@ -18,20 +18,6 @@
*/
package org.apache.pulsar.functions.utils;
-import com.google.protobuf.AbstractMessage.Builder;
-import com.google.protobuf.MessageOrBuilder;
-import com.google.protobuf.util.JsonFormat;
-import lombok.AccessLevel;
-import lombok.NoArgsConstructor;
-import lombok.extern.slf4j.Slf4j;
-import net.jodah.typetools.TypeResolver;
-import org.apache.pulsar.client.api.MessageId;
-import org.apache.pulsar.client.impl.MessageIdImpl;
-import org.apache.pulsar.functions.api.Function;
-import org.apache.pulsar.functions.proto.Function.FunctionDetails.Runtime;
-import org.apache.pulsar.io.core.Sink;
-import org.apache.pulsar.io.core.Source;
-
import java.io.File;
import java.io.IOException;
import java.lang.reflect.Constructor;
@@ -41,6 +27,23 @@
import java.net.ServerSocket;
import java.util.Collection;
+import org.apache.pulsar.client.api.MessageId;
+import org.apache.pulsar.client.impl.MessageIdImpl;
+import org.apache.pulsar.client.impl.TopicMessageIdImpl;
+import org.apache.pulsar.functions.api.Function;
+import org.apache.pulsar.functions.proto.Function.FunctionDetails.Runtime;
+import org.apache.pulsar.io.core.Sink;
+import org.apache.pulsar.io.core.Source;
+
+import com.google.protobuf.AbstractMessage.Builder;
+import com.google.protobuf.MessageOrBuilder;
+import com.google.protobuf.util.JsonFormat;
+
+import lombok.AccessLevel;
+import lombok.NoArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import net.jodah.typetools.TypeResolver;
+
/**
* Utils used for runtime.
*/
@@ -49,7 +52,9 @@
public class Utils {
public static final long getSequenceId(MessageId messageId) {
- MessageIdImpl msgId = (MessageIdImpl) messageId;
+ MessageIdImpl msgId = (MessageIdImpl) ((messageId instanceof TopicMessageIdImpl)
+ ? ((TopicMessageIdImpl) messageId).getInnerMessageId()
+ : messageId);
long ledgerId = msgId.getLedgerId();
long entryId = msgId.getEntryId();
diff --git a/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/validation/ValidatorImpls.java b/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/validation/ValidatorImpls.java
index 2ee6044c54..360289ce34 100644
--- a/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/validation/ValidatorImpls.java
+++ b/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/validation/ValidatorImpls.java
@@ -447,6 +447,10 @@ private static void doPythonChecks(FunctionConfig functionConfig, String name) {
if (functionConfig.getWindowConfig() != null) {
throw new IllegalArgumentException("There is currently no support windowing in python");
}
+
+ if (StringUtils.isNotBlank(functionConfig.getTopicsPattern())) {
+ throw new IllegalArgumentException("Topic-patterns is not supported for python runtime");
+ }
}
private static void verifyNoTopicClash(Collection<String> inputTopics, String outputTopic) throws IllegalArgumentException {
@@ -458,7 +462,8 @@ private static void verifyNoTopicClash(Collection<String> inputTopics, String ou
}
private static void doCommonChecks(FunctionConfig functionConfig) {
- if (functionConfig.getInputs().isEmpty() && functionConfig.getCustomSerdeInputs().isEmpty()) {
+ if ((functionConfig.getInputs().isEmpty() && StringUtils.isEmpty(functionConfig.getTopicsPattern()))
+ && functionConfig.getCustomSerdeInputs().isEmpty()) {
throw new RuntimeException("No input topic(s) specified for the function");
}
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services