You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by rd...@apache.org on 2018/06/07 05:00:08 UTC
[incubator-pulsar] branch master updated: Support pulsar source to
start consumer with topic patterns (#1903)
This is an automated email from the ASF dual-hosted git repository.
rdhabalia pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-pulsar.git
The following commit(s) were added to refs/heads/master by this push:
new 57f463d Support pulsar source to start consumer with topic patterns (#1903)
57f463d is described below
commit 57f463d11604533e3d3215baaf43e874a73da03d
Author: Rajan Dhabalia <rd...@apache.org>
AuthorDate: Wed Jun 6 22:00:01 2018 -0700
Support pulsar source to start consumer with topic patterns (#1903)
Pass sourceConfig.topicPattern from cmdSink
add test for topic-patterns
add topic-pattern to ProcessRunTime
document topicsPattern usage for java-func only
add python instance validaiton for topicPatterns args
getSequenceId for topicMessage
---
.../apache/pulsar/admin/cli/CmdFunctionsTest.java | 25 +++++++
.../org/apache/pulsar/admin/cli/CmdFunctions.java | 78 ++++++++++++----------
.../java/org/apache/pulsar/admin/cli/CmdSinks.java | 8 +++
.../functions/instance/JavaInstanceRunnable.java | 1 +
.../pulsar/functions/source/PulsarSource.java | 25 ++++++-
.../functions/source/PulsarSourceConfig.java | 1 +
.../src/main/python/python_instance_main.py | 3 +
.../pulsar/functions/source/PulsarSourceTest.java | 21 ++----
.../proto/src/main/proto/Function.proto | 1 +
.../pulsar/functions/runtime/JavaInstanceMain.java | 8 +++
.../pulsar/functions/runtime/ProcessRuntime.java | 6 ++
.../pulsar/functions/runtime/RuntimeSpawner.java | 10 +++
.../functions/runtime/ProcessRuntimeTest.java | 7 +-
.../pulsar/functions/utils/FunctionConfig.java | 2 +
.../apache/pulsar/functions/utils/SinkConfig.java | 3 +
.../org/apache/pulsar/functions/utils/Utils.java | 35 +++++-----
.../functions/utils/validation/ValidatorImpls.java | 7 +-
17 files changed, 169 insertions(+), 72 deletions(-)
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 20499e4..5861e22 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
@@ -206,6 +206,31 @@ public class CmdFunctionsTest {
}
@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";
String inputTopicName = "persistent://tenant/standalone/namespace/input-topic";
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 a200de8..6915774 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.utils.validation.ConfigValidation;
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 @@ public class CmdFunctions extends CmdBase {
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 @@ public class CmdFunctions extends CmdBase {
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 @@ public class CmdFunctions extends CmdBase {
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 14a6fb1..47f4ed7 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 @@ public class CmdSinks extends CmdBase {
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 @@ public class CmdSinks extends CmdBase {
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 @@ public class CmdSinks extends CmdBase {
});
}
sinkConfig.setTopicToSerdeClassName(topicsToSerDeClassName);
+
+ if (null != topicsPattern) {
+ sinkConfig.setTopicsPattern(topicsPattern);
+ }
if (parallelism != null) {
sinkConfig.setParallelism(parallelism);
@@ -299,6 +306,7 @@ public class CmdSinks extends CmdBase {
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 6bb689c..b79d508 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 class JavaInstanceRunnable implements AutoCloseable, Runnable {
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 adc28a1..74f2366 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 com.google.common.annotations.VisibleForTesting;
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 @@ public class PulsarSource<T> implements Source<T> {
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 class PulsarSource<T> implements Source<T> {
// 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 class PulsarSource<T> implements Source<T> {
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 6c45486..8f815f1 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 @@ public class PulsarSourceConfig {
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 ca490f8..b1ca25f 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 4c81016..3c5e61b 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 class PulsarSourceTest {
* 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 class PulsarSourceTest {
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 class PulsarSourceTest {
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 class PulsarSourceTest {
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 f9865c0..0f8b649 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 9a17567..b602a1d 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.Server;
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 @@ public class JavaInstanceMain implements AutoCloseable {
@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 class JavaInstanceMain implements AutoCloseable {
}
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 82b2d3b..715eab6 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.ManagedChannel;
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 @@ class ProcessRuntime implements Runtime {
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 60b41de..8b60ded 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 java.util.concurrent.ExecutionException;
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 class RuntimeSpawner implements AutoCloseable {
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 5a37328..65e8d89 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 @@ public class ProcessRuntimeTest {
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 class ProcessRuntimeTest {
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 class ProcessRuntimeTest {
+ " --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 class ProcessRuntimeTest {
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 class ProcessRuntimeTest {
+ " --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 9671d50..3140f95 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 @@ public class FunctionConfig {
@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 f4ee77c..097d626 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.
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 @@ public class SinkConfig {
@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 2601ca0..c5eba1f 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.lang.reflect.Type;
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 @@ import java.util.Collection;
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 2ee6044..360289c 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 @@ public class ValidatorImpls {
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 @@ public class ValidatorImpls {
}
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");
}
--
To stop receiving notification emails like this one, please contact
rdhabalia@apache.org.