From 2e84172181bf15881f21741af97c01bd7a4fe589 Mon Sep 17 00:00:00 2001 From: Matteo Merli Date: Tue, 9 Apr 2019 16:19:13 -0700 Subject: [PATCH 01/16] Allow to configure TypedMessageBuilder through a Map conf object --- .../api/SimpleTypedProducerConsumerTest.java | 80 +++++++++++++++++++ .../client/api/TypedMessageBuilder.java | 35 ++++++++ .../client/impl/TypedMessageBuilderImpl.java | 27 +++++++ .../impl/conf/ConfigurationDataUtils.java | 34 ++------ .../pulsar/client/util/TypeCheckUtil.java | 33 ++++++++ 5 files changed, 183 insertions(+), 26 deletions(-) create mode 100644 pulsar-client/src/main/java/org/apache/pulsar/client/util/TypeCheckUtil.java diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/SimpleTypedProducerConsumerTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/SimpleTypedProducerConsumerTest.java index 050db394fb0e5..c1ef47584343c 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/SimpleTypedProducerConsumerTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/SimpleTypedProducerConsumerTest.java @@ -18,15 +18,23 @@ */ package org.apache.pulsar.client.api; +import static org.testng.Assert.assertEquals; import static org.testng.Assert.fail; import com.google.common.base.MoreObjects; +import com.google.common.collect.Lists; import com.google.common.collect.Sets; + import java.time.Clock; import java.util.Collections; +import java.util.HashMap; +import java.util.Map; import java.util.Objects; import java.util.Set; import java.util.concurrent.TimeUnit; + +import lombok.Cleanup; + import org.apache.pulsar.broker.service.schema.SchemaCompatibilityStrategy; import org.apache.pulsar.broker.service.schema.SchemaRegistry; import org.apache.pulsar.client.api.schema.GenericRecord; @@ -623,4 +631,76 @@ public void testAutoBytesProducer() throws Exception { log.info("-- Exiting {} test --", methodName); } + + @Test + public void testMessageBuilderLoadConf() throws Exception { + String topic = "persistent://my-property/use/my-ns/my-topic-" + System.nanoTime(); + + @Cleanup + Consumer consumer = pulsarClient.newConsumer(Schema.STRING) + .topic(topic) + .subscriptionName("my-subscriber-name") + .subscribe(); + + @Cleanup + Producer producer = pulsarClient.newProducer(Schema.STRING) + .topic(topic) + .create(); + + Map properties = new HashMap<>(); + properties.put("a", "1"); + properties.put("b", "2"); + + Map msgConf = new HashMap<>(); + msgConf.put("key", "key-1"); + msgConf.put("properties", properties); + msgConf.put("eventTime", 1234); + msgConf.put("sequenceId", 5); + msgConf.put("replicationClusters", Lists.newArrayList("a", "b", "c")); + msgConf.put("disableReplication", false); + + producer.newMessage() + .value("my-message") + .loadConf(msgConf) + .send(); + + + Message msg = consumer.receive(); + assertEquals(msg.getKey(), "key-1"); + assertEquals(msg.getProperties().get("a"), "1"); + assertEquals(msg.getProperties().get("b"), "2"); + assertEquals(msg.getEventTime(), 1234); + assertEquals(msg.getSequenceId(), 5); + + consumer.acknowledge(msg); + + // Try with invalid confs + msgConf.clear(); + msgConf.put("nonExistingKey", "key-1"); + + try { + producer.newMessage() + .value("my-message") + .loadConf(msgConf) + .send(); + fail("Should have failed"); + } catch (RuntimeException e) { + // expected + } + + // Try with invalid type + msgConf.clear(); + msgConf.put("eventTime", "hello"); + + try { + producer.newMessage() + .value("my-message") + .loadConf(msgConf) + .send(); + fail("Should have failed"); + } catch (RuntimeException e) { + // expected + } + } + } diff --git a/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/TypedMessageBuilder.java b/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/TypedMessageBuilder.java index 423c08019d262..cb00f0138a85b 100644 --- a/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/TypedMessageBuilder.java +++ b/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/TypedMessageBuilder.java @@ -174,4 +174,39 @@ public interface TypedMessageBuilder extends Serializable { * @return the message builder instance */ TypedMessageBuilder disableReplication(); + + /** + * Configure the {@link TypedMessageBuilder} from a config map, as an alternative compared + * to call the individual builder methods. + *

+ * The "value" of the message itself cannot be set on the config map. + *

+ * Example: + * + * 

{@code
+     * Map conf = new HashMap<>();
+     * conf.put("key", "my-key");
+     * conf.put("eventTime", System.currentTimeMillis());
+     *
+     * producer.newMessage()
+     *             .value("my-message")
+     *             .loadConf(conf)
+     *             .send();
+     * }
+ * + * The available options are: + * + * + * + * + * + * + * + * + *
NameTypeDoc
{@code key}{@code String}{@link #key(String)}
{@code properties}{@code Map}{@link #properties(Map)}
{@code eventTime}{@code long}{@link #eventTime(long)}
{@code sequenceId}{@code long}{@link #sequenceId(long)}
{@code replicationClusters}{@code List}{@link #replicationClusters(List)}
{@code disableReplication}{@code boolean}{@link #disableReplication()}
+ * + * @param config a map with the configuration options for the message + * @return the message builder instance + */ + TypedMessageBuilder loadConf(Map config); } diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/TypedMessageBuilderImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/TypedMessageBuilderImpl.java index b13423ec1f448..1c483b9d1fa2c 100644 --- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/TypedMessageBuilderImpl.java +++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/TypedMessageBuilderImpl.java @@ -19,6 +19,7 @@ package org.apache.pulsar.client.impl; import static com.google.common.base.Preconditions.checkArgument; +import static org.apache.pulsar.client.util.TypeCheckUtil.checkType; import com.google.common.base.Preconditions; @@ -130,6 +131,32 @@ public TypedMessageBuilder disableReplication() { return this; } + @SuppressWarnings("unchecked") + @Override + public TypedMessageBuilder loadConf(Map config) { + config.forEach((key, value) -> { + if (key.equals("key")) { + this.key(checkType(value, String.class)); + } else if (key.equals("properties")) { + this.properties(checkType(value, Map.class)); + } else if (key.equals("eventTime")) { + this.eventTime(checkType(value, Number.class).longValue()); + } else if (key.equals("sequenceId")) { + this.sequenceId(checkType(value, Number.class).longValue()); + } else if (key.equals("replicationClusters")) { + this.replicationClusters(checkType(value, List.class)); + } else if (key.equals("disableReplication")) { + boolean disableReplication = checkType(value, Boolean.class); + if (disableReplication) { + this.disableReplication(); + } + } else { + throw new RuntimeException("Invalid message config key '" + key + "'"); + } + }); + return this; + } + public MessageMetadata.Builder getMetadataBuilder() { return msgMetadataBuilder; } diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/conf/ConfigurationDataUtils.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/conf/ConfigurationDataUtils.java index 4523939e07e56..20254b2705a9a 100644 --- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/conf/ConfigurationDataUtils.java +++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/conf/ConfigurationDataUtils.java @@ -18,45 +18,27 @@ */ package org.apache.pulsar.client.impl.conf; -import com.fasterxml.jackson.annotation.JsonInclude.Include; -import com.fasterxml.jackson.databind.DeserializationFeature; import com.fasterxml.jackson.databind.ObjectMapper; import com.google.common.collect.Maps; -import io.netty.util.concurrent.FastThreadLocal; + import java.io.IOException; import java.util.Map; +import lombok.experimental.UtilityClass; + +import org.apache.pulsar.common.util.ObjectMapperFactory; + /** * Utils for loading configuration data. */ +@UtilityClass public final class ConfigurationDataUtils { - public static ObjectMapper create() { - ObjectMapper mapper = new ObjectMapper(); - // forward compatibility for the properties may go away in the future - mapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, true); - mapper.configure(DeserializationFeature.READ_UNKNOWN_ENUM_VALUES_AS_NULL, false); - mapper.setSerializationInclusion(Include.NON_NULL); - return mapper; - } - - private static final FastThreadLocal mapper = new FastThreadLocal() { - @Override - protected ObjectMapper initialValue() throws Exception { - return create(); - } - }; - - public static ObjectMapper getThreadLocal() { - return mapper.get(); - } - - private ConfigurationDataUtils() {} - + @SuppressWarnings("unchecked") public static T loadData(Map config, T existingData, Class dataCls) { - ObjectMapper mapper = getThreadLocal(); + ObjectMapper mapper = ObjectMapperFactory.getThreadLocal(); try { String existingConfigJson = mapper.writeValueAsString(existingData); Map existingConfig = mapper.readValue(existingConfigJson, Map.class); diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/util/TypeCheckUtil.java b/pulsar-client/src/main/java/org/apache/pulsar/client/util/TypeCheckUtil.java new file mode 100644 index 0000000000000..494c964ed71ed --- /dev/null +++ b/pulsar-client/src/main/java/org/apache/pulsar/client/util/TypeCheckUtil.java @@ -0,0 +1,33 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.pulsar.client.util; + +import lombok.experimental.UtilityClass; + +@UtilityClass +public class TypeCheckUtil { + @SuppressWarnings("unchecked") + public static T checkType(Object o, Class clazz) { + if (!clazz.isInstance(o)) { + throw new RuntimeException( + String.format("Invalid object type '%s' when exepcting '%s'", o.getClass(), clazz)); + } + return (T) o; + } +} From 8885b0cbe78c4a6188c4c89892aa62c0f1a63310 Mon Sep 17 00:00:00 2001 From: Matteo Merli Date: Tue, 9 Apr 2019 16:51:33 -0700 Subject: [PATCH 02/16] Use constants for message confs --- .../api/SimpleTypedProducerConsumerTest.java | 2 +- .../client/api/TypedMessageBuilder.java | 21 ++++++++++++------- .../client/impl/TypedMessageBuilderImpl.java | 12 +++++------ 3 files changed, 21 insertions(+), 14 deletions(-) diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/SimpleTypedProducerConsumerTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/SimpleTypedProducerConsumerTest.java index c1ef47584343c..97438b7b3e5e3 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/SimpleTypedProducerConsumerTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/SimpleTypedProducerConsumerTest.java @@ -634,7 +634,7 @@ public void testAutoBytesProducer() throws Exception { @Test public void testMessageBuilderLoadConf() throws Exception { - String topic = "persistent://my-property/use/my-ns/my-topic-" + System.nanoTime(); + String topic = "my-topic-" + System.nanoTime(); @Cleanup Consumer consumer = pulsarClient.newConsumer(Schema.STRING) diff --git a/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/TypedMessageBuilder.java b/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/TypedMessageBuilder.java index cb00f0138a85b..a1e2f2da4f6aa 100644 --- a/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/TypedMessageBuilder.java +++ b/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/TypedMessageBuilder.java @@ -196,17 +196,24 @@ public interface TypedMessageBuilder extends Serializable { * * The available options are: * - * - * - * - * - * - * - * + * + * + * + * + * + * + * *
NameTypeDoc
{@code key}{@code String}{@link #key(String)}
{@code properties}{@code Map}{@link #properties(Map)}
{@code eventTime}{@code long}{@link #eventTime(long)}
{@code sequenceId}{@code long}{@link #sequenceId(long)}
{@code replicationClusters}{@code List}{@link #replicationClusters(List)}
{@code disableReplication}{@code boolean}{@link #disableReplication()}
ConstantNameTypeDoc
{@link #CONF_KEY}{@code key}{@code String}{@link #key(String)}
{@link #CONF_PROPERTIES}{@code properties}{@code Map}{@link #properties(Map)}
{@link #CONF_EVENT_TIME}{@code eventTime}{@code long}{@link #eventTime(long)}
{@link #CONF_SEQUENCE_ID}{@code sequenceId}{@code long}{@link #sequenceId(long)}
{@link #CONF_REPLICATION_CLUSTERS}{@code replicationClusters}{@code List}{@link #replicationClusters(List)}
{@link #CONF_DISABLE_REPLICATION}{@code disableReplication}{@code boolean}{@link #disableReplication()}
* * @param config a map with the configuration options for the message * @return the message builder instance */ TypedMessageBuilder loadConf(Map config); + + static final String CONF_KEY = "key"; + static final String CONF_PROPERTIES = "properties"; + static final String CONF_EVENT_TIME = "eventTime"; + static final String CONF_SEQUENCE_ID = "sequenceId"; + static final String CONF_REPLICATION_CLUSTERS = "replicationClusters"; + static final String CONF_DISABLE_REPLICATION = "disableReplication"; } diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/TypedMessageBuilderImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/TypedMessageBuilderImpl.java index 1c483b9d1fa2c..758c08ca673c6 100644 --- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/TypedMessageBuilderImpl.java +++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/TypedMessageBuilderImpl.java @@ -135,17 +135,17 @@ public TypedMessageBuilder disableReplication() { @Override public TypedMessageBuilder loadConf(Map config) { config.forEach((key, value) -> { - if (key.equals("key")) { + if (key.equals(CONF_KEY)) { this.key(checkType(value, String.class)); - } else if (key.equals("properties")) { + } else if (key.equals(CONF_PROPERTIES)) { this.properties(checkType(value, Map.class)); - } else if (key.equals("eventTime")) { + } else if (key.equals(CONF_EVENT_TIME)) { this.eventTime(checkType(value, Number.class).longValue()); - } else if (key.equals("sequenceId")) { + } else if (key.equals(CONF_SEQUENCE_ID)) { this.sequenceId(checkType(value, Number.class).longValue()); - } else if (key.equals("replicationClusters")) { + } else if (key.equals(CONF_REPLICATION_CLUSTERS)) { this.replicationClusters(checkType(value, List.class)); - } else if (key.equals("disableReplication")) { + } else if (key.equals(CONF_DISABLE_REPLICATION)) { boolean disableReplication = checkType(value, Boolean.class); if (disableReplication) { this.disableReplication(); From b5792b6448790772ed75a64b3d6865f924a8e525 Mon Sep 17 00:00:00 2001 From: Matteo Merli Date: Tue, 9 Apr 2019 17:33:17 -0700 Subject: [PATCH 03/16] Reverted previous change --- .../impl/conf/ConfigurationDataUtils.java | 34 ++++++++++++++----- 1 file changed, 26 insertions(+), 8 deletions(-) diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/conf/ConfigurationDataUtils.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/conf/ConfigurationDataUtils.java index 20254b2705a9a..4523939e07e56 100644 --- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/conf/ConfigurationDataUtils.java +++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/conf/ConfigurationDataUtils.java @@ -18,27 +18,45 @@ */ package org.apache.pulsar.client.impl.conf; +import com.fasterxml.jackson.annotation.JsonInclude.Include; +import com.fasterxml.jackson.databind.DeserializationFeature; import com.fasterxml.jackson.databind.ObjectMapper; import com.google.common.collect.Maps; - +import io.netty.util.concurrent.FastThreadLocal; import java.io.IOException; import java.util.Map; -import lombok.experimental.UtilityClass; - -import org.apache.pulsar.common.util.ObjectMapperFactory; - /** * Utils for loading configuration data. */ -@UtilityClass public final class ConfigurationDataUtils { - @SuppressWarnings("unchecked") + public static ObjectMapper create() { + ObjectMapper mapper = new ObjectMapper(); + // forward compatibility for the properties may go away in the future + mapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, true); + mapper.configure(DeserializationFeature.READ_UNKNOWN_ENUM_VALUES_AS_NULL, false); + mapper.setSerializationInclusion(Include.NON_NULL); + return mapper; + } + + private static final FastThreadLocal mapper = new FastThreadLocal() { + @Override + protected ObjectMapper initialValue() throws Exception { + return create(); + } + }; + + public static ObjectMapper getThreadLocal() { + return mapper.get(); + } + + private ConfigurationDataUtils() {} + public static T loadData(Map config, T existingData, Class dataCls) { - ObjectMapper mapper = ObjectMapperFactory.getThreadLocal(); + ObjectMapper mapper = getThreadLocal(); try { String existingConfigJson = mapper.writeValueAsString(existingData); Map existingConfig = mapper.readValue(existingConfigJson, Map.class); From 90aa99998afe3d94396e4257601f6d3b5e574dbd Mon Sep 17 00:00:00 2001 From: Matteo Merli Date: Tue, 9 Apr 2019 17:35:24 -0700 Subject: [PATCH 04/16] Use Long instead of Number --- .../pulsar/client/api/SimpleTypedProducerConsumerTest.java | 4 ++-- .../apache/pulsar/client/impl/TypedMessageBuilderImpl.java | 4 ++-- .../java/org/apache/pulsar/client/util/TypeCheckUtil.java | 2 +- 3 files changed, 5 insertions(+), 5 deletions(-) diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/SimpleTypedProducerConsumerTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/SimpleTypedProducerConsumerTest.java index 97438b7b3e5e3..586f72b49dec0 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/SimpleTypedProducerConsumerTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/SimpleTypedProducerConsumerTest.java @@ -654,8 +654,8 @@ public void testMessageBuilderLoadConf() throws Exception { Map msgConf = new HashMap<>(); msgConf.put("key", "key-1"); msgConf.put("properties", properties); - msgConf.put("eventTime", 1234); - msgConf.put("sequenceId", 5); + msgConf.put("eventTime", 1234L); + msgConf.put("sequenceId", 5L); msgConf.put("replicationClusters", Lists.newArrayList("a", "b", "c")); msgConf.put("disableReplication", false); diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/TypedMessageBuilderImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/TypedMessageBuilderImpl.java index 758c08ca673c6..fb18c0ec907eb 100644 --- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/TypedMessageBuilderImpl.java +++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/TypedMessageBuilderImpl.java @@ -140,9 +140,9 @@ public TypedMessageBuilder loadConf(Map config) { } else if (key.equals(CONF_PROPERTIES)) { this.properties(checkType(value, Map.class)); } else if (key.equals(CONF_EVENT_TIME)) { - this.eventTime(checkType(value, Number.class).longValue()); + this.eventTime(checkType(value, Long.class)); } else if (key.equals(CONF_SEQUENCE_ID)) { - this.sequenceId(checkType(value, Number.class).longValue()); + this.sequenceId(checkType(value, Long.class)); } else if (key.equals(CONF_REPLICATION_CLUSTERS)) { this.replicationClusters(checkType(value, List.class)); } else if (key.equals(CONF_DISABLE_REPLICATION)) { diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/util/TypeCheckUtil.java b/pulsar-client/src/main/java/org/apache/pulsar/client/util/TypeCheckUtil.java index 494c964ed71ed..cbabdfe1e5412 100644 --- a/pulsar-client/src/main/java/org/apache/pulsar/client/util/TypeCheckUtil.java +++ b/pulsar-client/src/main/java/org/apache/pulsar/client/util/TypeCheckUtil.java @@ -26,7 +26,7 @@ public class TypeCheckUtil { public static T checkType(Object o, Class clazz) { if (!clazz.isInstance(o)) { throw new RuntimeException( - String.format("Invalid object type '%s' when exepcting '%s'", o.getClass(), clazz)); + String.format("Invalid object type '%s' when exepcting '%s'", o.getClass().getName(), clazz.getName())); } return (T) o; } From e59e730fec6962e4e46505676a5a9875ff45896f Mon Sep 17 00:00:00 2001 From: Jerry Peng Date: Mon, 8 Apr 2019 12:36:59 -0700 Subject: [PATCH 05/16] Set key for message when using function publish --- pulsar-client-cpp/python/pulsar/functions/context.py | 3 ++- .../main/java/org/apache/pulsar/functions/api/Context.java | 2 ++ .../org/apache/pulsar/functions/instance/ContextImpl.java | 7 ++++++- pulsar-functions/instance/src/main/python/contextimpl.py | 4 +++- .../instance/src/main/python/python_instance.py | 2 +- 5 files changed, 14 insertions(+), 4 deletions(-) diff --git a/pulsar-client-cpp/python/pulsar/functions/context.py b/pulsar-client-cpp/python/pulsar/functions/context.py index 169ec7685821a..83989dd8ba34c 100644 --- a/pulsar-client-cpp/python/pulsar/functions/context.py +++ b/pulsar-client-cpp/python/pulsar/functions/context.py @@ -126,7 +126,8 @@ def record_metric(self, metric_name, metric_value): @abstractmethod def publish(self, topic_name, message, serde_class_name="serde.IdentitySerDe", properties=None, compression_type=None, callback=None): """Publishes message to topic_name by first serializing the message using serde_class_name serde - The message will have properties specified if any""" + The message will have properties specified if any + If input message has a key associated with it, the same key will be set by default for outgoing message """ pass @abstractmethod diff --git a/pulsar-functions/api-java/src/main/java/org/apache/pulsar/functions/api/Context.java b/pulsar-functions/api-java/src/main/java/org/apache/pulsar/functions/api/Context.java index 17f989e0ab5ee..a113807b8b356 100644 --- a/pulsar-functions/api-java/src/main/java/org/apache/pulsar/functions/api/Context.java +++ b/pulsar-functions/api-java/src/main/java/org/apache/pulsar/functions/api/Context.java @@ -224,6 +224,7 @@ public interface Context { /** * Publish an object using serDe for serializing to the topic. + * If input message has a key associated with it, the same key will be set by default for outgoing message * * @param topicName * The name of the topic for publishing @@ -237,6 +238,7 @@ public interface Context { /** * Publish an object to the topic using default schemas. + * If input message has a key associated with it, the same key will be set by default for outgoing message * * @param topicName The name of the topic for publishing * @param object The object that needs to be published diff --git a/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/ContextImpl.java b/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/ContextImpl.java index dc99f6074d05a..07b5ed1881b3a 100644 --- a/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/ContextImpl.java +++ b/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/ContextImpl.java @@ -32,6 +32,7 @@ import org.apache.pulsar.client.api.PulsarClient; import org.apache.pulsar.client.api.PulsarClientException; import org.apache.pulsar.client.api.Schema; +import org.apache.pulsar.client.api.TypedMessageBuilder; import org.apache.pulsar.client.impl.ProducerBuilderImpl; import org.apache.pulsar.common.util.FutureUtil; import org.apache.pulsar.functions.api.Context; @@ -382,7 +383,11 @@ public CompletableFuture publish(String topicName, O object, Schema } } - CompletableFuture future = producer.sendAsync(object).thenApply(msgId -> null); + TypedMessageBuilder messageBuilder = producer.newMessage(); + if (record.getKey().isPresent()) { + messageBuilder.key(record.getKey().get()); + } + CompletableFuture future = messageBuilder.value(object).sendAsync().thenApply(msgId -> null); future.exceptionally(e -> { this.statsManager.incrSysExceptions(e); logger.error("Failed to publish to topic {} with error {}", topicName, e); diff --git a/pulsar-functions/instance/src/main/python/contextimpl.py b/pulsar-functions/instance/src/main/python/contextimpl.py index f19381b2f3d82..35398d0db2ea9 100644 --- a/pulsar-functions/instance/src/main/python/contextimpl.py +++ b/pulsar-functions/instance/src/main/python/contextimpl.py @@ -172,7 +172,9 @@ def publish(self, topic_name, message, serde_class_name="serde.IdentitySerDe", p self.publish_serializers[serde_class_name] = serde_klass() output_bytes = bytes(self.publish_serializers[serde_class_name].serialize(message)) - self.publish_producers[topic_name].send_async(output_bytes, partial(self.callback_wrapper, callback, topic_name, self.get_message_id()), properties=properties) + self.publish_producers[topic_name].send_async( + output_bytes, partial(self.callback_wrapper, callback, topic_name, self.get_message_id()), + properties=properties, partition_key=self.message.partition_key()) def ack(self, msgid, topic): topic_consumer = None diff --git a/pulsar-functions/instance/src/main/python/python_instance.py b/pulsar-functions/instance/src/main/python/python_instance.py index 8f740f197c3de..13301f2b11276 100644 --- a/pulsar-functions/instance/src/main/python/python_instance.py +++ b/pulsar-functions/instance/src/main/python/python_instance.py @@ -284,7 +284,7 @@ def process_result(self, output, msg): if output_bytes is not None: props = {"__pfn_input_topic__" : str(msg.topic), "__pfn_input_msg_id__" : base64ify(msg.message.message_id().serialize())} - self.producer.send_async(output_bytes, partial(self.done_producing, msg.consumer, msg.message, self.producer.topic()), properties=props) + self.producer.send_async(output_bytes, partial(self.done_producing, msg.consumer, msg.message, self.producer.topic()), properties=props, partition_key=msg.message.partition_key()) elif self.auto_ack and self.atleast_once: msg.consumer.acknowledge(msg.message) From 7f0d7c5fe9a4afe8861aba0563651d0fb6971ced Mon Sep 17 00:00:00 2001 From: Jerry Peng Date: Mon, 8 Apr 2019 15:21:52 -0700 Subject: [PATCH 06/16] fix unit test --- .../functions/instance/ContextImplTest.java | 18 ++++++++++++++++++ 1 file changed, 18 insertions(+) diff --git a/pulsar-functions/instance/src/test/java/org/apache/pulsar/functions/instance/ContextImplTest.java b/pulsar-functions/instance/src/test/java/org/apache/pulsar/functions/instance/ContextImplTest.java index fd541f34fcdfd..9d2579c78b6af 100644 --- a/pulsar-functions/instance/src/test/java/org/apache/pulsar/functions/instance/ContextImplTest.java +++ b/pulsar-functions/instance/src/test/java/org/apache/pulsar/functions/instance/ContextImplTest.java @@ -21,9 +21,13 @@ import io.prometheus.client.CollectorRegistry; import org.apache.pulsar.client.api.Producer; import org.apache.pulsar.client.api.Schema; +import org.apache.pulsar.client.api.TypedMessageBuilder; +import org.apache.pulsar.client.impl.ProducerBase; import org.apache.pulsar.client.impl.ProducerBuilderImpl; import org.apache.pulsar.client.impl.PulsarClientImpl; +import org.apache.pulsar.client.impl.TypedMessageBuilderImpl; import org.apache.pulsar.client.impl.conf.ProducerConfigurationData; +import org.apache.pulsar.functions.api.Record; import org.apache.pulsar.functions.instance.state.StateContextImpl; import org.apache.pulsar.functions.proto.Function.FunctionDetails; import org.apache.pulsar.functions.secretsprovider.EnvironmentBasedSecretsProvider; @@ -38,10 +42,14 @@ import java.util.concurrent.CompletableFuture; import static java.nio.charset.StandardCharsets.UTF_8; +import static org.mockito.Matchers.any; import static org.mockito.Matchers.anyString; import static org.mockito.Matchers.eq; import static org.mockito.Matchers.same; +import static org.mockito.Mockito.doNothing; +import static org.mockito.Mockito.doReturn; import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.spy; import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; @@ -72,12 +80,22 @@ public void setup() { when(client.getSchema(anyString())).thenReturn(CompletableFuture.completedFuture(Optional.empty())); when(producer.sendAsync(anyString())).thenReturn(CompletableFuture.completedFuture(null)); + TypedMessageBuilder messageBuilder = spy(new TypedMessageBuilderImpl(mock(ProducerBase.class), Schema.STRING)); + doReturn(new CompletableFuture<>()).when(messageBuilder).sendAsync(); + when(producer.newMessage()).thenReturn(messageBuilder); + context = new ContextImpl( config, logger, client, new EnvironmentBasedSecretsProvider(), new CollectorRegistry(), new String[0], ComponentType.FUNCTION, null); + context.setCurrentMessageContext(new Record() { + @Override + public String getValue() { + return null; + } + }); } @Test(expectedExceptions = IllegalStateException.class) From 46ce38312b74e75a96954adf60f77bd4055794e1 Mon Sep 17 00:00:00 2001 From: Jerry Peng Date: Mon, 8 Apr 2019 15:59:51 -0700 Subject: [PATCH 07/16] fix python test --- .../instance/src/test/python/test_python_instance.py | 1 + 1 file changed, 1 insertion(+) diff --git a/pulsar-functions/instance/src/test/python/test_python_instance.py b/pulsar-functions/instance/src/test/python/test_python_instance.py index 8b92fa85f6b6c..bbd6ca1fe42f1 100644 --- a/pulsar-functions/instance/src/test/python/test_python_instance.py +++ b/pulsar-functions/instance/src/test/python/test_python_instance.py @@ -62,6 +62,7 @@ def test_context_publish(self): msg = Message() msg.message_id = Mock(return_value="test_message_id") + msg.partition_key = Mock(return_value="test_key") context_impl.set_current_message_context(msg, "test_topic_name") context_impl.publish("test_topic_name", "test_message") From 993c382e8a9c46b7ac3d78de34dadd972c8a4fa6 Mon Sep 17 00:00:00 2001 From: Jerry Peng Date: Mon, 8 Apr 2019 17:40:48 -0700 Subject: [PATCH 08/16] improving impl --- .../python/pulsar/functions/context.py | 8 +++- .../apache/pulsar/functions/api/Context.java | 43 +++++++++++++------ .../functions/instance/ContextImpl.java | 19 ++++++-- .../instance/src/main/python/contextimpl.py | 8 +++- 4 files changed, 58 insertions(+), 20 deletions(-) diff --git a/pulsar-client-cpp/python/pulsar/functions/context.py b/pulsar-client-cpp/python/pulsar/functions/context.py index 83989dd8ba34c..e3b3e97195b83 100644 --- a/pulsar-client-cpp/python/pulsar/functions/context.py +++ b/pulsar-client-cpp/python/pulsar/functions/context.py @@ -118,13 +118,19 @@ def get_secret(self, secret_name): """Returns the secret value associated with the name. None if nothing was found""" pass + @abstractmethod + def get_partition_key(self): + """Returns partition key of the input message is one exists""" + pass + + @abstractmethod def record_metric(self, metric_name, metric_value): """Records the metric_value. metric_value has to satisfy isinstance(metric_value, numbers.Number)""" pass @abstractmethod - def publish(self, topic_name, message, serde_class_name="serde.IdentitySerDe", properties=None, compression_type=None, callback=None): + def publish(self, topic_name, message, serde_class_name="serde.IdentitySerDe", properties=None, compression_type=None, callback=None, partition_key=None): """Publishes message to topic_name by first serializing the message using serde_class_name serde The message will have properties specified if any If input message has a key associated with it, the same key will be set by default for outgoing message """ diff --git a/pulsar-functions/api-java/src/main/java/org/apache/pulsar/functions/api/Context.java b/pulsar-functions/api-java/src/main/java/org/apache/pulsar/functions/api/Context.java index a113807b8b356..9433e1edb47b1 100644 --- a/pulsar-functions/api-java/src/main/java/org/apache/pulsar/functions/api/Context.java +++ b/pulsar-functions/api-java/src/main/java/org/apache/pulsar/functions/api/Context.java @@ -84,6 +84,7 @@ public interface Context { /** * The id of the function that we are executing + * * @return The function id */ String getFunctionId(); @@ -119,16 +120,22 @@ public interface Context { /** * Increment the builtin distributed counter referred by key. * - * @param key The name of the key + * @param key The name of the key * @param amount The amount to be incremented */ void incrCounter(String key, long amount); + /** + * Gets the partition key of the input message if there is one + * @return partition key + */ + Optional getPartitionKey(); + /** * Increment the builtin distributed counter referred by key * but dont wait for the completion of the increment operation * - * @param key The name of the key + * @param key The name of the key * @param amount The amount to be incremented */ CompletableFuture incrCounterAsync(String key, long amount); @@ -153,7 +160,7 @@ public interface Context { /** * Update the state value for the key. * - * @param key name of the key + * @param key name of the key * @param value state value of the key */ void putState(String key, ByteBuffer value); @@ -161,7 +168,7 @@ public interface Context { /** * Update the state value for the key, but don't wait for the operation to be completed * - * @param key name of the key + * @param key name of the key * @param value state value of the key */ CompletableFuture putStateAsync(String key, ByteBuffer value); @@ -218,20 +225,18 @@ public interface Context { * Record a user defined metric. * * @param metricName The name of the metric - * @param value The value of the metric + * @param value The value of the metric */ void recordMetric(String metricName, double value); /** - * Publish an object using serDe for serializing to the topic. + * Publish an object using serDe or schema class for serializing to the topic. * If input message has a key associated with it, the same key will be set by default for outgoing message * - * @param topicName - * The name of the topic for publishing - * @param object - * The object that needs to be published - * @param schemaOrSerdeClassName - * Either a builtin schema type (eg: "avro", "json", "protobuf") or the class name of the custom schema class + * @param topicName The name of the topic for publishing + * @param object The object that needs to be published + * @param schemaOrSerdeClassName Either a builtin schema type (eg: "avro", "json", "protobuf") or the class name + * of the custom schema class * @return A future that completes when the framework is done publishing the message */ CompletableFuture publish(String topicName, O object, String schemaOrSerdeClassName); @@ -241,9 +246,21 @@ public interface Context { * If input message has a key associated with it, the same key will be set by default for outgoing message * * @param topicName The name of the topic for publishing - * @param object The object that needs to be published + * @param object The object that needs to be published * @return A future that completes when the framework is done publishing the message */ CompletableFuture publish(String topicName, O object); + /** + * Publish an object using serDe or schema class for serializing to the topic. + * + * @param topicName The name of the topic for publishing + * @param object The object that needs to be published + * @param schemaOrSerdeClassName Either a builtin schema type (eg: "avro", "json", "protobuf") or the class name + * of the custom schema class + * @param partitionKey The message key to use when publishing to topic + * @return A future that completes when the framework is done publishing the message + */ + CompletableFuture publish(String topicName, O object, String schemaOrSerdeClassName, String partitionKey); + } \ No newline at end of file diff --git a/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/ContextImpl.java b/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/ContextImpl.java index 07b5ed1881b3a..f0dcdf4647e60 100644 --- a/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/ContextImpl.java +++ b/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/ContextImpl.java @@ -260,6 +260,11 @@ public String getSecret(String secretName) { } } + @Override + public Optional getPartitionKey() { + return record.getKey(); + } + private void ensureStateEnabled() { checkState(null != stateContext, "State is not enabled."); } @@ -337,11 +342,17 @@ public CompletableFuture publish(String topicName, O object) { @SuppressWarnings("unchecked") @Override public CompletableFuture publish(String topicName, O object, String schemaOrSerdeClassName) { - return publish(topicName, object, (Schema) topicSchema.getSchema(topicName, object, schemaOrSerdeClassName, false)); + return publish(topicName, object, schemaOrSerdeClassName, null); + } + + @SuppressWarnings("unchecked") + @Override + public CompletableFuture publish(String topicName, O object, String schemaOrSerdeClassName, String partitionKey) { + return publish(topicName, object, (Schema) topicSchema.getSchema(topicName, object, schemaOrSerdeClassName, false), Optional.ofNullable(partitionKey)); } @SuppressWarnings("unchecked") - public CompletableFuture publish(String topicName, O object, Schema schema) { + public CompletableFuture publish(String topicName, O object, Schema schema, Optional partitionKey) { Producer producer = (Producer) publishProducers.get(topicName); if (producer == null) { @@ -384,8 +395,8 @@ public CompletableFuture publish(String topicName, O object, Schema } TypedMessageBuilder messageBuilder = producer.newMessage(); - if (record.getKey().isPresent()) { - messageBuilder.key(record.getKey().get()); + if (partitionKey.isPresent()) { + messageBuilder.key(partitionKey.get()); } CompletableFuture future = messageBuilder.value(object).sendAsync().thenApply(msgId -> null); future.exceptionally(e -> { diff --git a/pulsar-functions/instance/src/main/python/contextimpl.py b/pulsar-functions/instance/src/main/python/contextimpl.py index 35398d0db2ea9..df4680b89018d 100644 --- a/pulsar-functions/instance/src/main/python/contextimpl.py +++ b/pulsar-functions/instance/src/main/python/contextimpl.py @@ -90,6 +90,9 @@ def get_message_properties(self): def get_current_message_topic_name(self): return self.message.topic_name() + def get_partition_key(self): + return self.message.partition_key() + def get_function_name(self): return self.instance_config.function_details.name @@ -146,7 +149,7 @@ def callback_wrapper(self, callback, topic, message_id, result, msg): if callback: callback(result, msg) - def publish(self, topic_name, message, serde_class_name="serde.IdentitySerDe", properties=None, compression_type=None, callback=None): + def publish(self, topic_name, message, serde_class_name="serde.IdentitySerDe", properties=None, compression_type=None, callback=None, partition_key=None): # Just make sure that user supplied values are properly typed topic_name = str(topic_name) serde_class_name = str(serde_class_name) @@ -172,9 +175,10 @@ def publish(self, topic_name, message, serde_class_name="serde.IdentitySerDe", p self.publish_serializers[serde_class_name] = serde_klass() output_bytes = bytes(self.publish_serializers[serde_class_name].serialize(message)) + self.publish_producers[topic_name].send_async( output_bytes, partial(self.callback_wrapper, callback, topic_name, self.get_message_id()), - properties=properties, partition_key=self.message.partition_key()) + properties=properties, partition_key=partition_key) def ack(self, msgid, topic): topic_consumer = None From ba9b884b361e10a6f96b6d9b7b99e317f3f5a318 Mon Sep 17 00:00:00 2001 From: Jerry Peng Date: Wed, 10 Apr 2019 17:53:58 -0700 Subject: [PATCH 09/16] improving implementation --- .../python/pulsar/functions/context.py | 26 +++++++++++++++++-- .../apache/pulsar/functions/api/Context.java | 13 ++++++++-- .../functions/instance/ContextImpl.java | 10 +++---- .../instance/src/main/python/contextimpl.py | 7 +++-- 4 files changed, 45 insertions(+), 11 deletions(-) diff --git a/pulsar-client-cpp/python/pulsar/functions/context.py b/pulsar-client-cpp/python/pulsar/functions/context.py index e3b3e97195b83..60b4707425044 100644 --- a/pulsar-client-cpp/python/pulsar/functions/context.py +++ b/pulsar-client-cpp/python/pulsar/functions/context.py @@ -130,12 +130,34 @@ def record_metric(self, metric_name, metric_value): pass @abstractmethod - def publish(self, topic_name, message, serde_class_name="serde.IdentitySerDe", properties=None, compression_type=None, callback=None, partition_key=None): - """Publishes message to topic_name by first serializing the message using serde_class_name serde + def publish(self, topic_name, message, serde_class_name="serde.IdentitySerDe", properties=None, compression_type=None, callback=None): + """ + + DEPRECATED + + Publishes message to topic_name by first serializing the message using serde_class_name serde The message will have properties specified if any If input message has a key associated with it, the same key will be set by default for outgoing message """ pass + @abstractmethod + def publish(self, topic_name, message, serde_class_name="serde.IdentitySerDe", compression_type=None, callback=None, message_conf=None): + """Publishes message to topic_name by first serializing the message using serde_class_name serde + The message will have properties specified if any + If input message has a key associated with it, the same key will be set by default for outgoing message + + The available options for message_conf: + + properties, + partition_key, + sequence_id, + replication_clusters, + disable_replication, + event_timestamp + + """ + pass + @abstractmethod def get_output_topic(self): """Returns the output topic of function""" diff --git a/pulsar-functions/api-java/src/main/java/org/apache/pulsar/functions/api/Context.java b/pulsar-functions/api-java/src/main/java/org/apache/pulsar/functions/api/Context.java index 9433e1edb47b1..28922d6d51f91 100644 --- a/pulsar-functions/api-java/src/main/java/org/apache/pulsar/functions/api/Context.java +++ b/pulsar-functions/api-java/src/main/java/org/apache/pulsar/functions/api/Context.java @@ -258,9 +258,18 @@ public interface Context { * @param object The object that needs to be published * @param schemaOrSerdeClassName Either a builtin schema type (eg: "avro", "json", "protobuf") or the class name * of the custom schema class - * @param partitionKey The message key to use when publishing to topic + * @param messageConf A map of configurations to set for the message that will be published + * The available options are: + * + * "key" - Parition Key + * "properties" - Map of properties + * "eventTime" + * "sequenceId" + * "replicationClusters" + * "disableReplication" + * * @return A future that completes when the framework is done publishing the message */ - CompletableFuture publish(String topicName, O object, String schemaOrSerdeClassName, String partitionKey); + CompletableFuture publish(String topicName, O object, String schemaOrSerdeClassName, Map messageConf); } \ No newline at end of file diff --git a/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/ContextImpl.java b/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/ContextImpl.java index f0dcdf4647e60..909b7985d6801 100644 --- a/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/ContextImpl.java +++ b/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/ContextImpl.java @@ -347,12 +347,12 @@ public CompletableFuture publish(String topicName, O object, String sc @SuppressWarnings("unchecked") @Override - public CompletableFuture publish(String topicName, O object, String schemaOrSerdeClassName, String partitionKey) { - return publish(topicName, object, (Schema) topicSchema.getSchema(topicName, object, schemaOrSerdeClassName, false), Optional.ofNullable(partitionKey)); + public CompletableFuture publish(String topicName, O object, String schemaOrSerdeClassName, Map messageProperties) { + return publish(topicName, object, (Schema) topicSchema.getSchema(topicName, object, schemaOrSerdeClassName, false), messageProperties); } @SuppressWarnings("unchecked") - public CompletableFuture publish(String topicName, O object, Schema schema, Optional partitionKey) { + public CompletableFuture publish(String topicName, O object, Schema schema, Map messageProperties) { Producer producer = (Producer) publishProducers.get(topicName); if (producer == null) { @@ -395,8 +395,8 @@ public CompletableFuture publish(String topicName, O object, Schema } TypedMessageBuilder messageBuilder = producer.newMessage(); - if (partitionKey.isPresent()) { - messageBuilder.key(partitionKey.get()); + if (messageBuilder != null) { + messageBuilder.loadConf(messageProperties); } CompletableFuture future = messageBuilder.value(object).sendAsync().thenApply(msgId -> null); future.exceptionally(e -> { diff --git a/pulsar-functions/instance/src/main/python/contextimpl.py b/pulsar-functions/instance/src/main/python/contextimpl.py index df4680b89018d..e13dcc8b92b3a 100644 --- a/pulsar-functions/instance/src/main/python/contextimpl.py +++ b/pulsar-functions/instance/src/main/python/contextimpl.py @@ -149,7 +149,10 @@ def callback_wrapper(self, callback, topic, message_id, result, msg): if callback: callback(result, msg) - def publish(self, topic_name, message, serde_class_name="serde.IdentitySerDe", properties=None, compression_type=None, callback=None, partition_key=None): + def publish(self, topic_name, message, serde_class_name="serde.IdentitySerDe", properties=None, compression_type=None, callback=None): + self.publish(topic_name, message, serde_class_name=serde_class_name, compression_type=compression_type, callback=callback, message_conf={"properties": properties}) + + def publish(self, topic_name, message, serde_class_name="serde.IdentitySerDe", compression_type=None, callback=None, message_conf=None): # Just make sure that user supplied values are properly typed topic_name = str(topic_name) serde_class_name = str(serde_class_name) @@ -178,7 +181,7 @@ def publish(self, topic_name, message, serde_class_name="serde.IdentitySerDe", p self.publish_producers[topic_name].send_async( output_bytes, partial(self.callback_wrapper, callback, topic_name, self.get_message_id()), - properties=properties, partition_key=partition_key) + **message_conf) def ack(self, msgid, topic): topic_consumer = None From 35a46409a89349cc534416ac1ae8624e904e23f3 Mon Sep 17 00:00:00 2001 From: Jerry Peng Date: Thu, 11 Apr 2019 14:31:37 -0700 Subject: [PATCH 10/16] add tests and examples --- .../worker/PulsarFunctionPublishTest.java | 385 ++++++++++++++++++ .../PublishFunctionWithMessageConf.java | 51 +++ .../publish_function_with_message_conf.py | 38 ++ .../functions/PulsarFunctionsTest.java | 160 +++++++- .../functions/PulsarFunctionsTestBase.java | 8 + 5 files changed, 632 insertions(+), 10 deletions(-) create mode 100644 pulsar-broker/src/test/java/org/apache/pulsar/functions/worker/PulsarFunctionPublishTest.java create mode 100644 pulsar-functions/java-examples/src/main/java/org/apache/pulsar/functions/api/examples/PublishFunctionWithMessageConf.java create mode 100644 pulsar-functions/python-examples/publish_function_with_message_conf.py diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/functions/worker/PulsarFunctionPublishTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/functions/worker/PulsarFunctionPublishTest.java new file mode 100644 index 0000000000000..ef88fddc7e156 --- /dev/null +++ b/pulsar-broker/src/test/java/org/apache/pulsar/functions/worker/PulsarFunctionPublishTest.java @@ -0,0 +1,385 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.pulsar.functions.worker; + +import com.google.common.collect.Lists; +import com.google.common.collect.Sets; +import org.apache.bookkeeper.test.PortManager; +import org.apache.pulsar.broker.PulsarService; +import org.apache.pulsar.broker.ServiceConfiguration; +import org.apache.pulsar.broker.ServiceConfigurationUtils; +import org.apache.pulsar.broker.authentication.AuthenticationProviderTls; +import org.apache.pulsar.broker.authorization.PulsarAuthorizationProvider; +import org.apache.pulsar.broker.loadbalance.impl.SimpleLoadManagerImpl; +import org.apache.pulsar.client.admin.BrokerStats; +import org.apache.pulsar.client.admin.PulsarAdmin; +import org.apache.pulsar.client.admin.PulsarAdminException; +import org.apache.pulsar.client.api.Authentication; +import org.apache.pulsar.client.api.ClientBuilder; +import org.apache.pulsar.client.api.Consumer; +import org.apache.pulsar.client.api.Message; +import org.apache.pulsar.client.api.Producer; +import org.apache.pulsar.client.api.PulsarClient; +import org.apache.pulsar.client.api.Schema; +import org.apache.pulsar.client.impl.auth.AuthenticationTls; +import org.apache.pulsar.common.functions.FunctionConfig; +import org.apache.pulsar.common.functions.Utils; +import org.apache.pulsar.common.naming.TopicName; +import org.apache.pulsar.common.policies.data.ClusterData; +import org.apache.pulsar.common.policies.data.FunctionStats; +import org.apache.pulsar.common.policies.data.SubscriptionStats; +import org.apache.pulsar.common.policies.data.TenantInfo; +import org.apache.pulsar.zookeeper.LocalBookkeeperEnsemble; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.testng.Assert; +import org.testng.annotations.AfterMethod; +import org.testng.annotations.BeforeMethod; +import org.testng.annotations.DataProvider; +import org.testng.annotations.Test; + +import java.io.File; +import java.io.FilenameFilter; +import java.lang.reflect.Method; +import java.net.URL; +import java.util.Arrays; +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.LinkedList; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.Set; +import java.util.concurrent.TimeUnit; + +import static org.apache.commons.lang3.StringUtils.isNotBlank; +import static org.apache.pulsar.broker.auth.MockedPulsarServiceBaseTest.retryStrategically; +import static org.apache.pulsar.functions.utils.functioncache.FunctionCacheEntry.JAVA_INSTANCE_JAR_PROPERTY; +import static org.mockito.Mockito.spy; +import static org.testng.Assert.assertEquals; +import static org.testng.Assert.assertNotEquals; + +/** + * Test Pulsar function state + * + */ +public class PulsarFunctionPublishTest { + LocalBookkeeperEnsemble bkEnsemble; + + ServiceConfiguration config; + WorkerConfig workerConfig; + URL urlTls; + PulsarService pulsar; + PulsarAdmin admin; + PulsarClient pulsarClient; + BrokerStats brokerStatsClient; + WorkerService functionsWorkerService; + final String tenant = "external-repl-prop"; + String pulsarFunctionsNamespace = tenant + "/use/pulsar-function-admin"; + String primaryHost; + String workerId; + + private final int ZOOKEEPER_PORT = PortManager.nextFreePort(); + private final List bookiePorts = new LinkedList<>(); + private final int brokerWebServicePort = PortManager.nextFreePort(); + private final int brokerWebServiceTlsPort = PortManager.nextFreePort(); + private final int brokerServicePort = PortManager.nextFreePort(); + private final int brokerServiceTlsPort = PortManager.nextFreePort(); + private final int workerServicePort = PortManager.nextFreePort(); + + private final String TLS_SERVER_CERT_FILE_PATH = "./src/test/resources/authentication/tls/broker-cert.pem"; + private final String TLS_SERVER_KEY_FILE_PATH = "./src/test/resources/authentication/tls/broker-key.pem"; + private final String TLS_CLIENT_CERT_FILE_PATH = "./src/test/resources/authentication/tls/client-cert.pem"; + private final String TLS_CLIENT_KEY_FILE_PATH = "./src/test/resources/authentication/tls/client-key.pem"; + private final String TLS_TRUST_CERT_FILE_PATH = "./src/test/resources/authentication/tls/cacert.pem"; + + private static final Logger log = LoggerFactory.getLogger(PulsarFunctionStateTest.class); + + @DataProvider(name = "validRoleName") + public Object[][] validRoleName() { + return new Object[][] { { Boolean.TRUE }, { Boolean.FALSE } }; + } + + @BeforeMethod + void setup(Method method) throws Exception { + + // delete all function temp files + File dir = new File(System.getProperty("java.io.tmpdir")); + File[] foundFiles = dir.listFiles(new FilenameFilter() { + public boolean accept(File dir, String name) { + return name.startsWith("function"); + } + }); + + for (File file : foundFiles) { + file.delete(); + } + + log.info("--- Setting up method {} ---", method.getName()); + + // Start local bookkeeper ensemble + bkEnsemble = new LocalBookkeeperEnsemble(3, ZOOKEEPER_PORT, () -> { + int port = PortManager.nextFreePort(); + bookiePorts.add(port); + return port; + }); + bkEnsemble.start(); + + String brokerServiceUrl = "https://127.0.0.1:" + brokerWebServiceTlsPort; + + config = spy(new ServiceConfiguration()); + config.setClusterName("use"); + Set superUsers = Sets.newHashSet("superUser"); + config.setSuperUserRoles(superUsers); + config.setWebServicePort(brokerWebServicePort); + config.setWebServicePortTls(brokerWebServiceTlsPort); + config.setZookeeperServers("127.0.0.1" + ":" + ZOOKEEPER_PORT); + config.setBrokerServicePort(brokerServicePort); + config.setBrokerServicePortTls(brokerServiceTlsPort); + config.setLoadManagerClassName(SimpleLoadManagerImpl.class.getName()); + config.setTlsAllowInsecureConnection(true); + config.setAdvertisedAddress("localhost"); + + Set providers = new HashSet<>(); + providers.add(AuthenticationProviderTls.class.getName()); + config.setAuthenticationEnabled(true); + config.setAuthenticationProviders(providers); + + config.setAuthorizationEnabled(true); + config.setAuthorizationProvider(PulsarAuthorizationProvider.class.getName()); + + config.setTlsCertificateFilePath(TLS_SERVER_CERT_FILE_PATH); + config.setTlsKeyFilePath(TLS_SERVER_KEY_FILE_PATH); + config.setTlsTrustCertsFilePath(TLS_TRUST_CERT_FILE_PATH); + + config.setBrokerClientAuthenticationPlugin(AuthenticationTls.class.getName()); + config.setBrokerClientAuthenticationParameters( + "tlsCertFile:" + TLS_CLIENT_CERT_FILE_PATH + "," + "tlsKeyFile:" + TLS_CLIENT_KEY_FILE_PATH); + config.setBrokerClientTrustCertsFilePath(TLS_TRUST_CERT_FILE_PATH); + config.setBrokerClientTlsEnabled(true); + + + + functionsWorkerService = createPulsarFunctionWorker(config); + urlTls = new URL(brokerServiceUrl); + Optional functionWorkerService = Optional.of(functionsWorkerService); + pulsar = new PulsarService(config, functionWorkerService); + pulsar.start(); + + Map authParams = new HashMap<>(); + authParams.put("tlsCertFile", TLS_CLIENT_CERT_FILE_PATH); + authParams.put("tlsKeyFile", TLS_CLIENT_KEY_FILE_PATH); + Authentication authTls = new AuthenticationTls(); + authTls.configure(authParams); + + admin = spy( + PulsarAdmin.builder().serviceHttpUrl(brokerServiceUrl).tlsTrustCertsFilePath(TLS_TRUST_CERT_FILE_PATH) + .allowTlsInsecureConnection(true).authentication(authTls).build()); + + brokerStatsClient = admin.brokerStats(); + primaryHost = String.format("http://%s:%d", "localhost", brokerWebServicePort); + + // update cluster metadata + ClusterData clusterData = new ClusterData(urlTls.toString()); + admin.clusters().updateCluster(config.getClusterName(), clusterData); + + ClientBuilder clientBuilder = PulsarClient.builder().serviceUrl(this.workerConfig.getPulsarServiceUrl()); + if (isNotBlank(workerConfig.getClientAuthenticationPlugin()) + && isNotBlank(workerConfig.getClientAuthenticationParameters())) { + clientBuilder.enableTls(workerConfig.isUseTls()); + clientBuilder.allowTlsInsecureConnection(workerConfig.isTlsAllowInsecureConnection()); + clientBuilder.authentication(workerConfig.getClientAuthenticationPlugin(), + workerConfig.getClientAuthenticationParameters()); + } + pulsarClient = clientBuilder.build(); + + TenantInfo propAdmin = new TenantInfo(); + propAdmin.getAdminRoles().add("superUser"); + propAdmin.setAllowedClusters(Sets.newHashSet(Lists.newArrayList("use"))); + admin.tenants().updateTenant(tenant, propAdmin); + + System.setProperty(JAVA_INSTANCE_JAR_PROPERTY, ""); + + } + + @AfterMethod + void shutdown() throws Exception { + log.info("--- Shutting down ---"); + pulsarClient.close(); + admin.close(); + functionsWorkerService.stop(); + pulsar.close(); + bkEnsemble.stop(); + } + + private WorkerService createPulsarFunctionWorker(ServiceConfiguration config) { + + workerConfig = new WorkerConfig(); + workerConfig.setPulsarFunctionsNamespace(pulsarFunctionsNamespace); + workerConfig.setSchedulerClassName( + org.apache.pulsar.functions.worker.scheduler.RoundRobinScheduler.class.getName()); + workerConfig.setThreadContainerFactory(new WorkerConfig.ThreadContainerFactory().setThreadGroupName("use")); +// workerConfig.setProcessContainerFactory(new WorkerConfig.ProcessContainerFactory() +// .setJavaInstanceJarLocation("/Users/jerrypeng/workspace/incubator-pulsar/pulsar-functions/runtime-all/target/java-instance.jar") +// .setPythonInstanceLocation("")); + // worker talks to local broker + workerConfig.setPulsarServiceUrl("pulsar://127.0.0.1:" + config.getBrokerServicePortTls().get()); + workerConfig.setPulsarWebServiceUrl("https://127.0.0.1:" + config.getWebServicePortTls().get()); + workerConfig.setFailureCheckFreqMs(100); + workerConfig.setNumFunctionPackageReplicas(1); + workerConfig.setClusterCoordinationTopicName("coordinate"); + workerConfig.setFunctionAssignmentTopicName("assignment"); + workerConfig.setFunctionMetadataTopicName("metadata"); + workerConfig.setInstanceLivenessCheckFreqMs(100); + workerConfig.setWorkerPort(workerServicePort); + workerConfig.setPulsarFunctionsCluster(config.getClusterName()); + String hostname = ServiceConfigurationUtils.getDefaultOrConfiguredAddress(config.getAdvertisedAddress()); + this.workerId = "c-" + config.getClusterName() + "-fw-" + hostname + "-" + workerConfig.getWorkerPort(); + workerConfig.setWorkerHostname(hostname); + workerConfig.setWorkerId(workerId); + + workerConfig.setClientAuthenticationPlugin(AuthenticationTls.class.getName()); + workerConfig.setClientAuthenticationParameters( + String.format("tlsCertFile:%s,tlsKeyFile:%s", TLS_CLIENT_CERT_FILE_PATH, TLS_CLIENT_KEY_FILE_PATH)); + workerConfig.setUseTls(true); + workerConfig.setTlsAllowInsecureConnection(true); + workerConfig.setTlsTrustCertsFilePath(TLS_TRUST_CERT_FILE_PATH); + + workerConfig.setAuthenticationEnabled(true); + workerConfig.setAuthorizationEnabled(true); + + return new WorkerService(workerConfig); + } + + protected static FunctionConfig createFunctionConfig(String tenant, String namespace, String functionName, String sourceTopic, String publishTopic, String subscriptionName) { + + FunctionConfig functionConfig = new FunctionConfig(); + functionConfig.setTenant(tenant); + functionConfig.setNamespace(namespace); + functionConfig.setName(functionName); + functionConfig.setParallelism(1); + functionConfig.setProcessingGuarantees(FunctionConfig.ProcessingGuarantees.EFFECTIVELY_ONCE); + functionConfig.setSubName(subscriptionName); + functionConfig.setInputs(Collections.singleton(sourceTopic)); + functionConfig.setAutoAck(true); + functionConfig.setClassName("org.apache.pulsar.functions.api.examples.PublishFunctionWithMessageConf"); + functionConfig.setRuntime(FunctionConfig.Runtime.JAVA); + Map userConfig = new HashMap<>(); + userConfig.put("publish-topic", publishTopic); + functionConfig.setUserConfig(userConfig); functionConfig.setCleanupSubscription(true); + return functionConfig; + } + + @Test(timeOut = 20000) + public void testPulsarFunctionState() throws Exception { + + final String namespacePortion = "io"; + final String replNamespace = tenant + "/" + namespacePortion; + final String sourceTopic = "persistent://" + replNamespace + "/input"; + final String publishTopic = "persistent://" + replNamespace + "/publishtopic"; + final String propertyKey = "key"; + final String propertyValue = "value"; + final String functionName = "PulsarFunction-test"; + final String subscriptionName = "test-sub"; + admin.namespaces().createNamespace(replNamespace); + Set clusters = Sets.newHashSet(Lists.newArrayList("use")); + admin.namespaces().setNamespaceReplicationClusters(replNamespace, clusters); + + // create a producer that creates a topic at broker + Producer producer = pulsarClient.newProducer(Schema.STRING).topic(sourceTopic).create(); + Consumer consumer = pulsarClient.newConsumer(Schema.STRING).topic(publishTopic).subscriptionName("sub").subscribe(); + + FunctionConfig functionConfig = createFunctionConfig(tenant, namespacePortion, functionName, + sourceTopic, publishTopic, subscriptionName); + + String jarFilePathUrl = Utils.FILE + ":" + getClass().getClassLoader().getResource("pulsar-functions-api-examples.jar").getFile(); + admin.functions().createFunctionWithUrl(functionConfig, jarFilePathUrl); + + retryStrategically((test) -> { + try { + return admin.topics().getStats(sourceTopic).subscriptions.size() == 1; + } catch (PulsarAdminException e) { + return false; + } + }, 5, 150); + // validate pulsar sink consumer has started on the topic + assertEquals(admin.topics().getStats(sourceTopic).subscriptions.size(), 1); + + int totalMsgs = 5; + for (int i = 0; i < totalMsgs; i++) { + String data = "foo"; + producer.newMessage().property(propertyKey, propertyValue).key(String.valueOf(i)).value(data).send(); + } + retryStrategically((test) -> { + try { + SubscriptionStats subStats = admin.topics().getStats(sourceTopic).subscriptions.get(subscriptionName); + return subStats.unackedMessages == 0; + } catch (PulsarAdminException e) { + return false; + } + }, 5, 150); + + retryStrategically((test) -> { + try { + FunctionStats functionStat = admin.functions().getFunctionStats(tenant, namespacePortion, functionName); + return functionStat.getProcessedSuccessfullyTotal() == 5; + } catch (PulsarAdminException e) { + return false; + } + }, 5, 150); + + for (int i = 0; i < 5; i++) { + Message msg = consumer.receive(5, TimeUnit.SECONDS); + String receivedPropertyValue = msg.getProperty(propertyKey); + assertEquals(propertyValue, receivedPropertyValue); + assertEquals(msg.getProperty("input_topic"), sourceTopic); + assertEquals(msg.getKey(), String.valueOf(i)); + } + + // validate pulsar-sink consumer has consumed all messages and delivered to Pulsar sink but unacked messages + // due to publish failure + assertNotEquals(admin.topics().getStats(sourceTopic).subscriptions.values().iterator().next().unackedMessages, + totalMsgs); + + // delete functions + admin.functions().deleteFunction(tenant, namespacePortion, functionName); + + retryStrategically((test) -> { + try { + return admin.topics().getStats(sourceTopic).subscriptions.size() == 0; + } catch (PulsarAdminException e) { + return false; + } + }, 5, 150); + + // make sure subscriptions are cleanup + assertEquals(admin.topics().getStats(sourceTopic).subscriptions.size(), 0); + + // make sure all temp files are deleted + File dir = new File(System.getProperty("java.io.tmpdir")); + File[] foundFiles = dir.listFiles(new FilenameFilter() { + public boolean accept(File dir, String name) { + return name.startsWith("function"); + } + }); + + Assert.assertEquals(foundFiles.length, 0, "Temporary files left over: " + Arrays.asList(foundFiles)); + } +} \ No newline at end of file diff --git a/pulsar-functions/java-examples/src/main/java/org/apache/pulsar/functions/api/examples/PublishFunctionWithMessageConf.java b/pulsar-functions/java-examples/src/main/java/org/apache/pulsar/functions/api/examples/PublishFunctionWithMessageConf.java new file mode 100644 index 0000000000000..ef69bac34d5b7 --- /dev/null +++ b/pulsar-functions/java-examples/src/main/java/org/apache/pulsar/functions/api/examples/PublishFunctionWithMessageConf.java @@ -0,0 +1,51 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.pulsar.functions.api.examples; + +import org.apache.pulsar.functions.api.Context; +import org.apache.pulsar.functions.api.Function; + +import java.util.HashMap; +import java.util.Map; + +/** + * Example function that uses the built in publish function in the context + * to publish to a desired topic based on config and setting various message configurations to be passed along. + * + */ +public class PublishFunctionWithMessageConf implements Function { + @Override + public Void process(String input, Context context) { + String publishTopic = (String) context.getUserConfigValueOrDefault("publish-topic", "publishtopic"); + String output = String.format("%s!", input); + + Map properties = new HashMap<>(); + properties.put("input_topic", context.getCurrentRecord().getTopicName().get()); + properties.putAll(context.getCurrentRecord().getProperties()); + + Map messageConf = new HashMap<>(); + messageConf.put("properties", properties); + if (context.getPartitionKey().isPresent()) { + messageConf.put("key", context.getPartitionKey().get()); + } + messageConf.put("eventTime", System.currentTimeMillis()); + context.publish(publishTopic, output, null, messageConf); + return null; + } +} diff --git a/pulsar-functions/python-examples/publish_function_with_message_conf.py b/pulsar-functions/python-examples/publish_function_with_message_conf.py new file mode 100644 index 0000000000000..79aac0239fa55 --- /dev/null +++ b/pulsar-functions/python-examples/publish_function_with_message_conf.py @@ -0,0 +1,38 @@ +#!/usr/bin/env python +# +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +# + +import time +from pulsar import Function + +# Example function that uses the built in publish function in the context +# to publish to a desired topic based on config +class PublishFunctionWithMessageConf(Function): + def __init__(self): + pass + + def process(self, input, context): + publish_topic = "publishtopic" + if "publish-topic" in context.get_user_config_map(): + publish_topic = context.get_user_config_value("publish-topic") + context.publish(publish_topic, input + '!', + message_conf={"properties": {k: v for d in [{"input_topic" : context.get_current_message_topic_name()}, context.get_message_properties()] for k, v in d.items()}, + "partition_key": context.get_partition_key(), + "event_timestamp": int(time.time())}) + return diff --git a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/functions/PulsarFunctionsTest.java b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/functions/PulsarFunctionsTest.java index a60eb47ebfb12..bb4c6b4cfe454 100644 --- a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/functions/PulsarFunctionsTest.java +++ b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/functions/PulsarFunctionsTest.java @@ -26,6 +26,7 @@ import com.google.common.base.Stopwatch; import com.google.gson.Gson; +import java.util.Collections; import java.util.HashSet; import java.util.LinkedHashMap; import java.util.Map; @@ -724,6 +725,108 @@ protected void getSourceInfoNotFound(String tenant, String namespace, String sou // Test CRUD functions on different runtimes. // + @Test + public void testPythonPublishFunction() throws Exception { + testPublishFunction(Runtime.PYTHON); + } + + @Test + public void testJavaPublishFunction() throws Exception { + testPublishFunction(Runtime.JAVA); + } + + private void testPublishFunction(Runtime runtime) throws Exception { + if (functionRuntimeType == FunctionRuntimeType.THREAD) { + return; + } + + Schema schema; + if (Runtime.JAVA == runtime) { + schema = Schema.STRING; + } else { + schema = Schema.BYTES; + } + + String inputTopicName = "persistent://public/default/test-publish-" + runtime + "-input-" + randomName(8); + String outputTopicName = "test-publish-" + runtime + "-output-" + randomName(8); + + String functionName = "test-publish-fn-" + randomName(8); + final int numMessages = 10; + + // submit the exclamation function + + if (runtime == Runtime.PYTHON) { + submitFunction( + runtime, inputTopicName, outputTopicName, functionName, PUBLISH_FUNCTION_PYTHON_FILE, PUBLISH_PYTHON_CLASS, schema, Collections.singletonMap("publish-topic", outputTopicName)); + } else { + submitFunction( + runtime, inputTopicName, outputTopicName, functionName, null, PUBLISH_JAVA_CLASS, schema, Collections.singletonMap("publish-topic", outputTopicName)); + } + + // get function info + getFunctionInfoSuccess(functionName); + + // get function stats + getFunctionStatsEmpty(functionName); + + // publish and consume result + if (Runtime.JAVA == runtime) { + // java supports schema + publishAndConsumeMessages(inputTopicName, outputTopicName, numMessages); + } else { + // python doesn't support schema + + @Cleanup PulsarClient client = PulsarClient.builder() + .serviceUrl(pulsarCluster.getPlainTextServiceUrl()) + .build(); + @Cleanup Consumer consumer = client.newConsumer(Schema.BYTES) + .topic(outputTopicName) + .subscriptionType(SubscriptionType.Exclusive) + .subscriptionName("test-sub") + .subscribe(); + + @Cleanup Producer producer = client.newProducer(Schema.BYTES) + .topic(inputTopicName) + .create(); + + for (int i = 0; i < numMessages; i++) { + producer.newMessage().key(String.valueOf(i)).property("count", String.valueOf(i)).value(("message-" + i).getBytes(UTF_8)).send(); + } + + Set expectedMessages = new HashSet<>(); + for (int i = 0; i < numMessages; i++) { + expectedMessages.add("message-" + i + "!"); + } + + for (int i = 0; i < numMessages; i++) { + Message msg = consumer.receive(30, TimeUnit.SECONDS); + String msgValue = new String(msg.getValue(), UTF_8); + log.info("Received: {}", msgValue); + assertEquals(msg.getKey(), String.valueOf(i)); + assertEquals(msg.getProperties().get("count"), String.valueOf(i)); + assertEquals(msg.getProperties().get("input_topic"), inputTopicName); + assertTrue(msg.getEventTime() > 0); + assertTrue(expectedMessages.contains(msgValue)); + expectedMessages.remove(msgValue); + } + } + + // get function status + getFunctionStatus(functionName, numMessages, true); + + // get function stats + getFunctionStats(functionName, numMessages); + + // delete function + deleteFunction(functionName); + + // get function info + getFunctionInfoNotFound(functionName); + + // make sure subscriptions are cleanup + checkSubscriptionsCleanup(inputTopicName); + } + @Test public void testPythonExclamationFunction() throws Exception { testExclamationFunction(Runtime.PYTHON, false, false, false); @@ -841,6 +944,7 @@ private static void submitExclamationFunction(Runtime runtime, functionName, pyZip, withExtraDeps, + false, getExclamationClass(runtime, pyZip, withExtraDeps), schema); } @@ -851,8 +955,47 @@ private static void submitFunction(Runtime runtime, String functionName, boolean pyZip, boolean withExtraDeps, + boolean isPublishFunction, + String functionClass, + Schema inputTopicSchema) throws Exception { + + String file = null; + if (Runtime.JAVA == runtime) { + file = null; + } else if (Runtime.PYTHON == runtime) { + if (isPublishFunction) { + file = PUBLISH_FUNCTION_PYTHON_FILE; + } else if (pyZip) { + file = EXCLAMATION_PYTHONZIP_FILE; + } else if (withExtraDeps) { + file = EXCLAMATION_WITH_DEPS_PYTHON_FILE; + } else { + file = EXCLAMATION_PYTHON_FILE; + } + } + + submitFunction(runtime, inputTopicName, outputTopicName, functionName, file, functionClass, inputTopicSchema); + } + + private static void submitFunction(Runtime runtime, + String inputTopicName, + String outputTopicName, + String functionName, + String functionFile, String functionClass, Schema inputTopicSchema) throws Exception { + submitFunction(runtime, inputTopicName, outputTopicName, functionName, functionFile, functionClass, inputTopicSchema, null); + } + + private static void submitFunction(Runtime runtime, + String inputTopicName, + String outputTopicName, + String functionName, + String functionFile, + String functionClass, + Schema inputTopicSchema, + Map userConfigs) throws Exception { + CommandGenerator generator; log.info("------- INPUT TOPIC: '{}'", inputTopicName); if (inputTopicName.endsWith(".*")) { @@ -864,28 +1007,25 @@ private static void submitFunction(Runtime runtime, } generator.setSinkTopic(outputTopicName); generator.setFunctionName(functionName); + if (userConfigs != null) { + generator.setUserConfig(userConfigs); + } String command; if (Runtime.JAVA == runtime) { command = generator.generateCreateFunctionCommand(); } else if (Runtime.PYTHON == runtime) { generator.setRuntime(runtime); - if (pyZip) { - command = generator.generateCreateFunctionCommand(EXCLAMATION_PYTHONZIP_FILE); - } else if (withExtraDeps) { - command = generator.generateCreateFunctionCommand(EXCLAMATION_WITH_DEPS_PYTHON_FILE); - } else { - command = generator.generateCreateFunctionCommand(EXCLAMATION_PYTHON_FILE); - } + command = generator.generateCreateFunctionCommand(functionFile); } else { throw new IllegalArgumentException("Unsupported runtime : " + runtime); } log.info("---------- Function command: {}", command); String[] commands = { - "sh", "-c", command + "sh", "-c", command }; ContainerExecResult result = pulsarCluster.getAnyWorker().execCmd( - commands); + commands); assertTrue(result.getStdout().contains("\"Created successfully\"")); ensureSubscriptionCreated(inputTopicName, String.format("public/default/%s", functionName), inputTopicSchema); @@ -1181,7 +1321,7 @@ public void testAutoSchemaFunction() throws Exception { // submit the exclamation function submitFunction( - Runtime.JAVA, inputTopicName, outputTopicName, functionName, false, false, + Runtime.JAVA, inputTopicName, outputTopicName, functionName, false, false, false, AutoSchemaFunction.class.getName(), Schema.AVRO(CustomObject.class)); diff --git a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/functions/PulsarFunctionsTestBase.java b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/functions/PulsarFunctionsTestBase.java index 851793c56dcb5..e7173896868e9 100644 --- a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/functions/PulsarFunctionsTestBase.java +++ b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/functions/PulsarFunctionsTestBase.java @@ -73,6 +73,10 @@ public void teardownFunctionWorkers() { public static final String EXCLAMATION_JAVA_CLASS = "org.apache.pulsar.functions.api.examples.ExclamationFunction"; + public static final String PUBLISH_JAVA_CLASS = + "org.apache.pulsar.functions.api.examples.PublishFunctionWithMessageConf"; + + public static final String EXCLAMATION_PYTHON_CLASS = "exclamation_function.ExclamationFunction"; @@ -82,9 +86,13 @@ public void teardownFunctionWorkers() { public static final String EXCLAMATION_PYTHONZIP_CLASS = "exclamation"; + public static final String PUBLISH_PYTHON_CLASS = "publish_function_with_message_conf.PublishFunctionWithMessageConf"; + public static final String EXCLAMATION_PYTHON_FILE = "exclamation_function.py"; public static final String EXCLAMATION_WITH_DEPS_PYTHON_FILE = "exclamation_with_extra_deps.py"; public static final String EXCLAMATION_PYTHONZIP_FILE = "exclamation.zip"; + public static final String PUBLISH_FUNCTION_PYTHON_FILE = "publish_function_with_message_conf.py"; + protected static String getExclamationClass(Runtime runtime, boolean pyZip, From 45f79de8c038af25c54b58ce34ebfc4f068938b7 Mon Sep 17 00:00:00 2001 From: Jerry Peng Date: Thu, 11 Apr 2019 14:49:11 -0700 Subject: [PATCH 11/16] fix bug --- .../apache/pulsar/functions/instance/ContextImpl.java | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/ContextImpl.java b/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/ContextImpl.java index 909b7985d6801..c95536d1c45f6 100644 --- a/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/ContextImpl.java +++ b/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/ContextImpl.java @@ -347,12 +347,12 @@ public CompletableFuture publish(String topicName, O object, String sc @SuppressWarnings("unchecked") @Override - public CompletableFuture publish(String topicName, O object, String schemaOrSerdeClassName, Map messageProperties) { - return publish(topicName, object, (Schema) topicSchema.getSchema(topicName, object, schemaOrSerdeClassName, false), messageProperties); + public CompletableFuture publish(String topicName, O object, String schemaOrSerdeClassName, Map messageConf) { + return publish(topicName, object, (Schema) topicSchema.getSchema(topicName, object, schemaOrSerdeClassName, false), messageConf); } @SuppressWarnings("unchecked") - public CompletableFuture publish(String topicName, O object, Schema schema, Map messageProperties) { + public CompletableFuture publish(String topicName, O object, Schema schema, Map messageConf) { Producer producer = (Producer) publishProducers.get(topicName); if (producer == null) { @@ -395,8 +395,8 @@ public CompletableFuture publish(String topicName, O object, Schema } TypedMessageBuilder messageBuilder = producer.newMessage(); - if (messageBuilder != null) { - messageBuilder.loadConf(messageProperties); + if (messageConf != null) { + messageBuilder.loadConf(messageConf); } CompletableFuture future = messageBuilder.value(object).sendAsync().thenApply(msgId -> null); future.exceptionally(e -> { From 03be81350f1fb727f1cf4811ef8e7cc7b7f26a75 Mon Sep 17 00:00:00 2001 From: Jerry Peng Date: Thu, 11 Apr 2019 15:55:03 -0700 Subject: [PATCH 12/16] fix bug --- pulsar-functions/instance/src/main/python/contextimpl.py | 9 ++++++--- 1 file changed, 6 insertions(+), 3 deletions(-) diff --git a/pulsar-functions/instance/src/main/python/contextimpl.py b/pulsar-functions/instance/src/main/python/contextimpl.py index e13dcc8b92b3a..fbba3d5a93415 100644 --- a/pulsar-functions/instance/src/main/python/contextimpl.py +++ b/pulsar-functions/instance/src/main/python/contextimpl.py @@ -179,9 +179,12 @@ def publish(self, topic_name, message, serde_class_name="serde.IdentitySerDe", c output_bytes = bytes(self.publish_serializers[serde_class_name].serialize(message)) - self.publish_producers[topic_name].send_async( - output_bytes, partial(self.callback_wrapper, callback, topic_name, self.get_message_id()), - **message_conf) + if message_conf: + self.publish_producers[topic_name].send_async( + output_bytes, partial(self.callback_wrapper, callback, topic_name, self.get_message_id()), **message_conf) + else: + self.publish_producers[topic_name].send_async( + output_bytes, partial(self.callback_wrapper, callback, topic_name, self.get_message_id())) def ack(self, msgid, topic): topic_consumer = None From 34361aebc3416507451a8b27735b1a271a24176e Mon Sep 17 00:00:00 2001 From: Jerry Peng Date: Fri, 12 Apr 2019 15:17:14 -0700 Subject: [PATCH 13/16] fixing comments --- pulsar-client-cpp/python/pulsar/functions/context.py | 3 +-- .../main/java/org/apache/pulsar/functions/api/Context.java | 7 ------- .../org/apache/pulsar/functions/instance/ContextImpl.java | 7 +------ 3 files changed, 2 insertions(+), 15 deletions(-) diff --git a/pulsar-client-cpp/python/pulsar/functions/context.py b/pulsar-client-cpp/python/pulsar/functions/context.py index 60b4707425044..14b277a021bc2 100644 --- a/pulsar-client-cpp/python/pulsar/functions/context.py +++ b/pulsar-client-cpp/python/pulsar/functions/context.py @@ -137,14 +137,13 @@ def publish(self, topic_name, message, serde_class_name="serde.IdentitySerDe", p Publishes message to topic_name by first serializing the message using serde_class_name serde The message will have properties specified if any - If input message has a key associated with it, the same key will be set by default for outgoing message """ + """ pass @abstractmethod def publish(self, topic_name, message, serde_class_name="serde.IdentitySerDe", compression_type=None, callback=None, message_conf=None): """Publishes message to topic_name by first serializing the message using serde_class_name serde The message will have properties specified if any - If input message has a key associated with it, the same key will be set by default for outgoing message The available options for message_conf: diff --git a/pulsar-functions/api-java/src/main/java/org/apache/pulsar/functions/api/Context.java b/pulsar-functions/api-java/src/main/java/org/apache/pulsar/functions/api/Context.java index 28922d6d51f91..556086a271877 100644 --- a/pulsar-functions/api-java/src/main/java/org/apache/pulsar/functions/api/Context.java +++ b/pulsar-functions/api-java/src/main/java/org/apache/pulsar/functions/api/Context.java @@ -125,11 +125,6 @@ public interface Context { */ void incrCounter(String key, long amount); - /** - * Gets the partition key of the input message if there is one - * @return partition key - */ - Optional getPartitionKey(); /** * Increment the builtin distributed counter referred by key @@ -231,7 +226,6 @@ public interface Context { /** * Publish an object using serDe or schema class for serializing to the topic. - * If input message has a key associated with it, the same key will be set by default for outgoing message * * @param topicName The name of the topic for publishing * @param object The object that needs to be published @@ -243,7 +237,6 @@ public interface Context { /** * Publish an object to the topic using default schemas. - * If input message has a key associated with it, the same key will be set by default for outgoing message * * @param topicName The name of the topic for publishing * @param object The object that needs to be published diff --git a/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/ContextImpl.java b/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/ContextImpl.java index c95536d1c45f6..5c06b498c58f8 100644 --- a/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/ContextImpl.java +++ b/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/ContextImpl.java @@ -259,12 +259,7 @@ public String getSecret(String secretName) { return null; } } - - @Override - public Optional getPartitionKey() { - return record.getKey(); - } - + private void ensureStateEnabled() { checkState(null != stateContext, "State is not enabled."); } From bcae09c14f73ce765acd26736aa19ab2b8c7e19a Mon Sep 17 00:00:00 2001 From: Jerry Peng Date: Fri, 12 Apr 2019 15:18:40 -0700 Subject: [PATCH 14/16] fix example --- .../api/examples/PublishFunctionWithMessageConf.java | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/pulsar-functions/java-examples/src/main/java/org/apache/pulsar/functions/api/examples/PublishFunctionWithMessageConf.java b/pulsar-functions/java-examples/src/main/java/org/apache/pulsar/functions/api/examples/PublishFunctionWithMessageConf.java index ef69bac34d5b7..be79deff2ad64 100644 --- a/pulsar-functions/java-examples/src/main/java/org/apache/pulsar/functions/api/examples/PublishFunctionWithMessageConf.java +++ b/pulsar-functions/java-examples/src/main/java/org/apache/pulsar/functions/api/examples/PublishFunctionWithMessageConf.java @@ -41,8 +41,8 @@ public Void process(String input, Context context) { Map messageConf = new HashMap<>(); messageConf.put("properties", properties); - if (context.getPartitionKey().isPresent()) { - messageConf.put("key", context.getPartitionKey().get()); + if (context.getCurrentRecord().getKey().isPresent()) { + messageConf.put("key", context.getCurrentRecord().getKey().get()); } messageConf.put("eventTime", System.currentTimeMillis()); context.publish(publishTopic, output, null, messageConf); From d58330e1eb08f4e7beaf4c6b95e753abceb743e2 Mon Sep 17 00:00:00 2001 From: Jerry Peng Date: Fri, 12 Apr 2019 18:28:39 -0700 Subject: [PATCH 15/16] addressing comments --- .../instance/src/main/python/python_instance.py | 2 +- .../api/examples/PublishFunctionWithMessageConf.java | 7 ++++--- 2 files changed, 5 insertions(+), 4 deletions(-) diff --git a/pulsar-functions/instance/src/main/python/python_instance.py b/pulsar-functions/instance/src/main/python/python_instance.py index 13301f2b11276..8f740f197c3de 100644 --- a/pulsar-functions/instance/src/main/python/python_instance.py +++ b/pulsar-functions/instance/src/main/python/python_instance.py @@ -284,7 +284,7 @@ def process_result(self, output, msg): if output_bytes is not None: props = {"__pfn_input_topic__" : str(msg.topic), "__pfn_input_msg_id__" : base64ify(msg.message.message_id().serialize())} - self.producer.send_async(output_bytes, partial(self.done_producing, msg.consumer, msg.message, self.producer.topic()), properties=props, partition_key=msg.message.partition_key()) + self.producer.send_async(output_bytes, partial(self.done_producing, msg.consumer, msg.message, self.producer.topic()), properties=props) elif self.auto_ack and self.atleast_once: msg.consumer.acknowledge(msg.message) diff --git a/pulsar-functions/java-examples/src/main/java/org/apache/pulsar/functions/api/examples/PublishFunctionWithMessageConf.java b/pulsar-functions/java-examples/src/main/java/org/apache/pulsar/functions/api/examples/PublishFunctionWithMessageConf.java index be79deff2ad64..7abb73ba421a8 100644 --- a/pulsar-functions/java-examples/src/main/java/org/apache/pulsar/functions/api/examples/PublishFunctionWithMessageConf.java +++ b/pulsar-functions/java-examples/src/main/java/org/apache/pulsar/functions/api/examples/PublishFunctionWithMessageConf.java @@ -18,6 +18,7 @@ */ package org.apache.pulsar.functions.api.examples; +import org.apache.pulsar.client.api.TypedMessageBuilder; import org.apache.pulsar.functions.api.Context; import org.apache.pulsar.functions.api.Function; @@ -40,11 +41,11 @@ public Void process(String input, Context context) { properties.putAll(context.getCurrentRecord().getProperties()); Map messageConf = new HashMap<>(); - messageConf.put("properties", properties); + messageConf.put(TypedMessageBuilder.CONF_PROPERTIES, properties); if (context.getCurrentRecord().getKey().isPresent()) { - messageConf.put("key", context.getCurrentRecord().getKey().get()); + messageConf.put(TypedMessageBuilder.CONF_KEY, context.getCurrentRecord().getKey().get()); } - messageConf.put("eventTime", System.currentTimeMillis()); + messageConf.put(TypedMessageBuilder.CONF_KEY, System.currentTimeMillis()); context.publish(publishTopic, output, null, messageConf); return null; } From 332405d3942e911d9c1dab1c28cb6c0a96947a42 Mon Sep 17 00:00:00 2001 From: Jerry Peng Date: Fri, 12 Apr 2019 22:01:27 -0700 Subject: [PATCH 16/16] fix function --- .../functions/api/examples/PublishFunctionWithMessageConf.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pulsar-functions/java-examples/src/main/java/org/apache/pulsar/functions/api/examples/PublishFunctionWithMessageConf.java b/pulsar-functions/java-examples/src/main/java/org/apache/pulsar/functions/api/examples/PublishFunctionWithMessageConf.java index 7abb73ba421a8..9960552cbee69 100644 --- a/pulsar-functions/java-examples/src/main/java/org/apache/pulsar/functions/api/examples/PublishFunctionWithMessageConf.java +++ b/pulsar-functions/java-examples/src/main/java/org/apache/pulsar/functions/api/examples/PublishFunctionWithMessageConf.java @@ -45,7 +45,7 @@ public Void process(String input, Context context) { if (context.getCurrentRecord().getKey().isPresent()) { messageConf.put(TypedMessageBuilder.CONF_KEY, context.getCurrentRecord().getKey().get()); } - messageConf.put(TypedMessageBuilder.CONF_KEY, System.currentTimeMillis()); + messageConf.put(TypedMessageBuilder.CONF_EVENT_TIME, System.currentTimeMillis()); context.publish(publishTopic, output, null, messageConf); return null; }