From c786346f74cfae4796f2e352145b7aa6bd4e46c3 Mon Sep 17 00:00:00 2001 From: "xiaolong.ran" Date: Sat, 20 Apr 2019 14:36:58 +0800 Subject: [PATCH 01/16] issue:4042 improve java functions API Signed-off-by: xiaolong.ran --- pulsar-functions/api-java/pom.xml | 6 ++ .../apache/pulsar/functions/api/Context.java | 7 +++ .../functions/instance/ContextImpl.java | 57 ++++++++++++++++++- 3 files changed, 69 insertions(+), 1 deletion(-) diff --git a/pulsar-functions/api-java/pom.xml b/pulsar-functions/api-java/pom.xml index 6f0481db9c43b..fe6554b708bf9 100644 --- a/pulsar-functions/api-java/pom.xml +++ b/pulsar-functions/api-java/pom.xml @@ -42,6 +42,12 @@ typetools test + + org.apache.pulsar + pulsar-client-api + 2.4.0-SNAPSHOT + compile + 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 556086a271877..28f7cf6bec65a 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 @@ -19,6 +19,9 @@ package org.apache.pulsar.functions.api; import java.nio.ByteBuffer; + +import org.apache.pulsar.client.api.Schema; +import org.apache.pulsar.client.api.TypedMessageBuilder; import org.slf4j.Logger; import java.util.Collection; @@ -265,4 +268,8 @@ public interface Context { */ CompletableFuture publish(String topicName, O object, String schemaOrSerdeClassName, Map messageConf); + TypedMessageBuilder newOutputMessage(String topicName, Schema schema); + + CompletableFuture publish(String topicName, O object, String schemaOrSerdeClassName, TypedMessageBuilder messageBuilder); + } \ 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 5c06b498c58f8..0b17e9e38b1df 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 @@ -337,7 +337,7 @@ public CompletableFuture publish(String topicName, O object) { @SuppressWarnings("unchecked") @Override public CompletableFuture publish(String topicName, O object, String schemaOrSerdeClassName) { - return publish(topicName, object, schemaOrSerdeClassName, null); + return publish(topicName, object, schemaOrSerdeClassName, (Map) null); } @SuppressWarnings("unchecked") @@ -346,6 +346,61 @@ public CompletableFuture publish(String topicName, O object, String sc return publish(topicName, object, (Schema) topicSchema.getSchema(topicName, object, schemaOrSerdeClassName, false), messageConf); } + @Override + public TypedMessageBuilder newOutputMessage(String topicName, Schema schema) { + Producer producer = (Producer) publishProducers.get(topicName); + if (producer == null) { + try { + Producer newProducer = ((ProducerBuilderImpl) producerBuilder.clone()) + .schema(schema) + .blockIfQueueFull(true) + .enableBatching(true) + .batchingMaxPublishDelay(10, TimeUnit.MILLISECONDS) + .compressionType(CompressionType.LZ4) + .hashingScheme(HashingScheme.Murmur3_32Hash) // + .messageRoutingMode(MessageRoutingMode.CustomPartition) + .messageRouter(FunctionResultRouter.of()) + // set send timeout to be infinity to prevent potential deadlock with consumer + // that might happen when consumer is blocked due to unacked messages + .sendTimeout(0, TimeUnit.SECONDS) + .topic(topicName) + .properties(InstanceUtils.getProperties(componentType, + FunctionCommon.getFullyQualifiedName( + this.config.getFunctionDetails().getTenant(), + this.config.getFunctionDetails().getNamespace(), + this.config.getFunctionDetails().getName()), + this.config.getInstanceId())) + .create(); + + Producer existingProducer = (Producer) publishProducers.putIfAbsent(topicName, newProducer); + + if (existingProducer != null) { + // The value in the map was not updated after the concurrent put + newProducer.close(); + producer = existingProducer; + } else { + producer = newProducer; + } + + } catch (PulsarClientException e) { + logger.error("Failed to create Producer while doing user publish", e); + return (TypedMessageBuilder) FutureUtil.failedFuture(e); + } + } + return producer.newMessage(); + } + + @Override + public CompletableFuture publish(String topicName, O object, String schemaOrSerdeClassName, TypedMessageBuilder messageBuilder) { + 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); + return null; + }); + return future; + } + @SuppressWarnings("unchecked") public CompletableFuture publish(String topicName, O object, Schema schema, Map messageConf) { Producer producer = (Producer) publishProducers.get(topicName); From 5b593f3a639f3d6308e1a8de8bd1492ebe574590 Mon Sep 17 00:00:00 2001 From: "xiaolong.ran" Date: Sat, 20 Apr 2019 15:42:28 +0800 Subject: [PATCH 02/16] fix code logic Signed-off-by: xiaolong.ran --- .../apache/pulsar/functions/api/Context.java | 6 +- .../functions/instance/ContextImpl.java | 159 +++++++----------- 2 files changed, 60 insertions(+), 105 deletions(-) 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 28f7cf6bec65a..bad269b33e34b 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 @@ -20,6 +20,7 @@ import java.nio.ByteBuffer; +import org.apache.pulsar.client.api.PulsarClientException; import org.apache.pulsar.client.api.Schema; import org.apache.pulsar.client.api.TypedMessageBuilder; import org.slf4j.Logger; @@ -268,8 +269,5 @@ public interface Context { */ CompletableFuture publish(String topicName, O object, String schemaOrSerdeClassName, Map messageConf); - TypedMessageBuilder newOutputMessage(String topicName, Schema schema); - - CompletableFuture publish(String topicName, O object, String schemaOrSerdeClassName, TypedMessageBuilder messageBuilder); - + TypedMessageBuilder newOutputMessage(String topicName, Schema schema) throws PulsarClientException; } \ 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 0b17e9e38b1df..886c118560176 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,114 +347,31 @@ public CompletableFuture publish(String topicName, O object, String sc } @Override - public TypedMessageBuilder newOutputMessage(String topicName, Schema schema) { - Producer producer = (Producer) publishProducers.get(topicName); - if (producer == null) { - try { - Producer newProducer = ((ProducerBuilderImpl) producerBuilder.clone()) - .schema(schema) - .blockIfQueueFull(true) - .enableBatching(true) - .batchingMaxPublishDelay(10, TimeUnit.MILLISECONDS) - .compressionType(CompressionType.LZ4) - .hashingScheme(HashingScheme.Murmur3_32Hash) // - .messageRoutingMode(MessageRoutingMode.CustomPartition) - .messageRouter(FunctionResultRouter.of()) - // set send timeout to be infinity to prevent potential deadlock with consumer - // that might happen when consumer is blocked due to unacked messages - .sendTimeout(0, TimeUnit.SECONDS) - .topic(topicName) - .properties(InstanceUtils.getProperties(componentType, - FunctionCommon.getFullyQualifiedName( - this.config.getFunctionDetails().getTenant(), - this.config.getFunctionDetails().getNamespace(), - this.config.getFunctionDetails().getName()), - this.config.getInstanceId())) - .create(); - - Producer existingProducer = (Producer) publishProducers.putIfAbsent(topicName, newProducer); - - if (existingProducer != null) { - // The value in the map was not updated after the concurrent put - newProducer.close(); - producer = existingProducer; - } else { - producer = newProducer; - } - - } catch (PulsarClientException e) { - logger.error("Failed to create Producer while doing user publish", e); - return (TypedMessageBuilder) FutureUtil.failedFuture(e); - } - } + public TypedMessageBuilder newOutputMessage(String topicName, Schema schema) throws PulsarClientException { + Producer producer = getProducer(topicName, schema); return producer.newMessage(); } - @Override - public CompletableFuture publish(String topicName, O object, String schemaOrSerdeClassName, TypedMessageBuilder messageBuilder) { - 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); - return null; - }); - return future; - } - @SuppressWarnings("unchecked") public CompletableFuture publish(String topicName, O object, Schema schema, Map messageConf) { - Producer producer = (Producer) publishProducers.get(topicName); - - if (producer == null) { - try { - Producer newProducer = ((ProducerBuilderImpl) producerBuilder.clone()) - .schema(schema) - .blockIfQueueFull(true) - .enableBatching(true) - .batchingMaxPublishDelay(10, TimeUnit.MILLISECONDS) - .compressionType(CompressionType.LZ4) - .hashingScheme(HashingScheme.Murmur3_32Hash) // - .messageRoutingMode(MessageRoutingMode.CustomPartition) - .messageRouter(FunctionResultRouter.of()) - // set send timeout to be infinity to prevent potential deadlock with consumer - // that might happen when consumer is blocked due to unacked messages - .sendTimeout(0, TimeUnit.SECONDS) - .topic(topicName) - .properties(InstanceUtils.getProperties(componentType, - FunctionCommon.getFullyQualifiedName( - this.config.getFunctionDetails().getTenant(), - this.config.getFunctionDetails().getNamespace(), - this.config.getFunctionDetails().getName()), - this.config.getInstanceId())) - .create(); - - Producer existingProducer = (Producer) publishProducers.putIfAbsent(topicName, newProducer); - - if (existingProducer != null) { - // The value in the map was not updated after the concurrent put - newProducer.close(); - producer = existingProducer; - } else { - producer = newProducer; - } - - } catch (PulsarClientException e) { - logger.error("Failed to create Producer while doing user publish", e); - return FutureUtil.failedFuture(e); + Producer producer; + try { + producer = getProducer(topicName, schema); + TypedMessageBuilder messageBuilder = producer.newMessage(); + if (messageConf != null) { + messageBuilder.loadConf(messageConf); } + 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); + return null; + }); + return future; + } catch (PulsarClientException e) { + logger.error("Failed to create Producer while doing user publish", e); + return FutureUtil.failedFuture(e); } - - TypedMessageBuilder messageBuilder = producer.newMessage(); - if (messageConf != null) { - messageBuilder.loadConf(messageConf); - } - 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); - return null; - }); - return future; } @Override @@ -472,6 +389,46 @@ public void recordMetric(String metricName, double value) { } } + private Producer getProducer(String topicName, Schema schema) throws PulsarClientException { + Producer producer = (Producer) publishProducers.get(topicName); + + if (producer == null) { + + Producer newProducer = ((ProducerBuilderImpl) producerBuilder.clone()) + .schema(schema) + .blockIfQueueFull(true) + .enableBatching(true) + .batchingMaxPublishDelay(10, TimeUnit.MILLISECONDS) + .compressionType(CompressionType.LZ4) + .hashingScheme(HashingScheme.Murmur3_32Hash) // + .messageRoutingMode(MessageRoutingMode.CustomPartition) + .messageRouter(FunctionResultRouter.of()) + // set send timeout to be infinity to prevent potential deadlock with consumer + // that might happen when consumer is blocked due to unacked messages + .sendTimeout(0, TimeUnit.SECONDS) + .topic(topicName) + .properties(InstanceUtils.getProperties(componentType, + FunctionCommon.getFullyQualifiedName( + this.config.getFunctionDetails().getTenant(), + this.config.getFunctionDetails().getNamespace(), + this.config.getFunctionDetails().getName()), + this.config.getInstanceId())) + .create(); + + Producer existingProducer = (Producer) publishProducers.putIfAbsent(topicName, newProducer); + + if (existingProducer != null) { + // The value in the map was not updated after the concurrent put + newProducer.close(); + producer = existingProducer; + } else { + producer = newProducer; + } + + } + return producer; + } + public Map getAndResetMetrics() { Map retval = getMetrics(); resetMetrics(); From 83d778e67d81dbfb01ec6d25efab92706bdd9b12 Mon Sep 17 00:00:00 2001 From: "xiaolong.ran" Date: Sat, 20 Apr 2019 16:10:03 +0800 Subject: [PATCH 03/16] change publish methods to use newOutputMessage Signed-off-by: xiaolong.ran --- .../java/org/apache/pulsar/functions/instance/ContextImpl.java | 3 +-- 1 file changed, 1 insertion(+), 2 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 886c118560176..7d5c608a6b0da 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 @@ -356,8 +356,7 @@ public TypedMessageBuilder newOutputMessage(String topicName, Schema s public CompletableFuture publish(String topicName, O object, Schema schema, Map messageConf) { Producer producer; try { - producer = getProducer(topicName, schema); - TypedMessageBuilder messageBuilder = producer.newMessage(); + TypedMessageBuilder messageBuilder = newOutputMessage(topicName, schema); if (messageConf != null) { messageBuilder.loadConf(messageConf); } From 2e455adbd3e867e5e1395725a4f8395b7d5ee370 Mon Sep 17 00:00:00 2001 From: "xiaolong.ran" Date: Mon, 22 Apr 2019 22:25:32 +0800 Subject: [PATCH 04/16] add java doc for newOutputMessage Signed-off-by: xiaolong.ran --- .../apache/pulsar/functions/api/Context.java | 12 ++ .../functions/instance/ContextImpl.java | 122 ++++++++++++++---- 2 files changed, 110 insertions(+), 24 deletions(-) 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 bad269b33e34b..85472003def00 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 @@ -236,6 +236,7 @@ public interface Context { * @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 + * @deprecated in favor of using {@link #newOutputMessage(String, Schema)} */ CompletableFuture publish(String topicName, O object, String schemaOrSerdeClassName); @@ -245,6 +246,7 @@ public interface Context { * @param topicName The name of the topic for publishing * @param object The object that needs to be published * @return A future that completes when the framework is done publishing the message + * @deprecated in favor of using {@link #newOutputMessage(String, Schema)} */ CompletableFuture publish(String topicName, O object); @@ -266,8 +268,18 @@ public interface Context { * "disableReplication" * * @return A future that completes when the framework is done publishing the message + * @deprecated in favor of using {@link #newOutputMessage(String, Schema)} */ CompletableFuture publish(String topicName, O object, String schemaOrSerdeClassName, Map messageConf); + /** + * New output message using schema for serializing to the topic + * + * @param topicName The name of the topic for output message + * @param schema provide a way to convert between serialized data and domain objects + * @param + * @return the message builder instance + * @throws PulsarClientException + */ TypedMessageBuilder newOutputMessage(String topicName, Schema schema) throws PulsarClientException; } \ 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 7d5c608a6b0da..cb380cca6e65b 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 @@ -25,14 +25,7 @@ import lombok.Getter; import lombok.Setter; import org.apache.commons.lang3.StringUtils; -import org.apache.pulsar.client.api.CompressionType; -import org.apache.pulsar.client.api.HashingScheme; -import org.apache.pulsar.client.api.MessageRoutingMode; -import org.apache.pulsar.client.api.Producer; -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.api.*; import org.apache.pulsar.client.impl.ProducerBuilderImpl; import org.apache.pulsar.common.util.FutureUtil; import org.apache.pulsar.functions.api.Context; @@ -52,12 +45,9 @@ import org.slf4j.Logger; import java.nio.ByteBuffer; -import java.util.Arrays; -import java.util.Collection; -import java.util.HashMap; -import java.util.Map; -import java.util.Optional; +import java.util.*; import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ExecutionException; import java.util.concurrent.TimeUnit; import static com.google.common.base.Preconditions.checkState; @@ -94,11 +84,13 @@ class ContextImpl implements Context, SinkContext, SourceContext { private final Summary userMetricsSummary; private final static String[] userMetricsLabelNames; + static { // add label to indicate user metric userMetricsLabelNames = Arrays.copyOf(ComponentStatsManager.metricsLabelNames, ComponentStatsManager.metricsLabelNames.length + 1); userMetricsLabelNames[ComponentStatsManager.metricsLabelNames.length] = "metric"; } + private final ComponentType componentType; public ContextImpl(InstanceConfig config, Logger logger, PulsarClient client, @@ -259,7 +251,7 @@ public String getSecret(String secretName) { return null; } } - + private void ensureStateEnabled() { checkState(null != stateContext, "State is not enabled."); } @@ -348,25 +340,107 @@ public CompletableFuture publish(String topicName, O object, String sc @Override public TypedMessageBuilder newOutputMessage(String topicName, Schema schema) throws PulsarClientException { - Producer producer = getProducer(topicName, schema); - return producer.newMessage(); + final TypedMessageBuilder underlyingBuilder = getProducer(topicName, schema).newMessage(); + return new TypedMessageBuilder() { + + @Override + public MessageId send() throws PulsarClientException { + try { + return sendAsync().get(); + } catch (ExecutionException e) { + Throwable t = e.getCause(); + if (t instanceof PulsarClientException) { + throw (PulsarClientException) t; + } else { + throw new PulsarClientException(t); + } + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new PulsarClientException(e); + } + } + + @Override + public CompletableFuture sendAsync() { + return underlyingBuilder.sendAsync() + .whenComplete((result, cause) -> { + if (null != cause) { + statsManager.incrSysExceptions(cause); + logger.error("Failed to publish to topic {} with error {}", topicName, cause); + } + }); + } + + @Override + public TypedMessageBuilder key(String key) { + underlyingBuilder.key(key); + return this; + } + + @Override + public TypedMessageBuilder keyBytes(byte[] key) { + underlyingBuilder.keyBytes(key); + return this; + } + + @Override + public TypedMessageBuilder value(O value) { + underlyingBuilder.value(value); + return this; + } + + @Override + public TypedMessageBuilder property(String name, String value) { + underlyingBuilder.property(name, value); + return this; + } + + @Override + public TypedMessageBuilder properties(Map properties) { + underlyingBuilder.properties(properties); + return this; + } + + @Override + public TypedMessageBuilder eventTime(long timestamp) { + underlyingBuilder.eventTime(timestamp); + return this; + } + + @Override + public TypedMessageBuilder sequenceId(long sequenceId) { + underlyingBuilder.sequenceId(sequenceId); + return this; + } + + @Override + public TypedMessageBuilder replicationClusters(List clusters) { + underlyingBuilder.replicationClusters(clusters); + return this; + } + + @Override + public TypedMessageBuilder disableReplication() { + underlyingBuilder.disableReplication(); + return this; + } + + @Override + public TypedMessageBuilder loadConf(Map config) { + underlyingBuilder.loadConf(config); + return this; + } + }; } @SuppressWarnings("unchecked") public CompletableFuture publish(String topicName, O object, Schema schema, Map messageConf) { - Producer producer; try { TypedMessageBuilder messageBuilder = newOutputMessage(topicName, schema); if (messageConf != null) { messageBuilder.loadConf(messageConf); } - 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); - return null; - }); - return future; + return messageBuilder.value(object).sendAsync().thenApply(msgId -> null); } catch (PulsarClientException e) { logger.error("Failed to create Producer while doing user publish", e); return FutureUtil.failedFuture(e); From 5b0bfe9bafa7edec06b0c6421d4c1d662228672f Mon Sep 17 00:00:00 2001 From: "xiaolong.ran" Date: Tue, 23 Apr 2019 10:11:41 +0800 Subject: [PATCH 05/16] update publish function examples Signed-off-by: xiaolong.ran --- .../functions/api/examples/PublishFunction.java | 9 ++++++++- .../examples/PublishFunctionWithMessageConf.java | 14 ++++++++------ 2 files changed, 16 insertions(+), 7 deletions(-) diff --git a/pulsar-functions/java-examples/src/main/java/org/apache/pulsar/functions/api/examples/PublishFunction.java b/pulsar-functions/java-examples/src/main/java/org/apache/pulsar/functions/api/examples/PublishFunction.java index cda83313c80ed..0c7cb56ea825c 100644 --- a/pulsar-functions/java-examples/src/main/java/org/apache/pulsar/functions/api/examples/PublishFunction.java +++ b/pulsar-functions/java-examples/src/main/java/org/apache/pulsar/functions/api/examples/PublishFunction.java @@ -18,6 +18,8 @@ */ package org.apache.pulsar.functions.api.examples; +import org.apache.pulsar.client.api.PulsarClientException; +import org.apache.pulsar.client.api.TypedMessageBuilder; import org.apache.pulsar.functions.api.Context; import org.apache.pulsar.functions.api.Function; @@ -30,7 +32,12 @@ public class PublishFunction implements Function { public Void process(String input, Context context) { String publishTopic = (String) context.getUserConfigValueOrDefault("publish-topic", "publishtopic"); String output = String.format("%s!", input); - context.publish(publishTopic, output); + try { + TypedMessageBuilder msgBuilder = context.newOutputMessage(publishTopic, null); + msgBuilder.value(output).sendAsync(); + } catch (PulsarClientException e) { + e.printStackTrace(); + } return null; } } 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 9960552cbee69..bb034b24f3383 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.PulsarClientException; import org.apache.pulsar.client.api.TypedMessageBuilder; import org.apache.pulsar.functions.api.Context; import org.apache.pulsar.functions.api.Function; @@ -40,13 +41,14 @@ public Void process(String input, Context context) { properties.put("input_topic", context.getCurrentRecord().getTopicName().get()); properties.putAll(context.getCurrentRecord().getProperties()); - Map messageConf = new HashMap<>(); - messageConf.put(TypedMessageBuilder.CONF_PROPERTIES, properties); - if (context.getCurrentRecord().getKey().isPresent()) { - messageConf.put(TypedMessageBuilder.CONF_KEY, context.getCurrentRecord().getKey().get()); + try { + TypedMessageBuilder msgBuilder = context.newOutputMessage(publishTopic, null); + msgBuilder.value(output).properties(properties). + key(context.getCurrentRecord().getKey().get()). + eventTime(System.currentTimeMillis()).sendAsync(); + } catch (PulsarClientException e) { + e.printStackTrace(); } - messageConf.put(TypedMessageBuilder.CONF_EVENT_TIME, System.currentTimeMillis()); - context.publish(publishTopic, output, null, messageConf); return null; } } From 22bc0f7757552058ac1ec2fc60b3caf2b53f38c0 Mon Sep 17 00:00:00 2001 From: "xiaolong.ran" Date: Tue, 23 Apr 2019 21:37:36 +0800 Subject: [PATCH 06/16] remove pulish() function Signed-off-by: xiaolong.ran --- .../apache/pulsar/functions/api/Context.java | 22 -- .../functions/instance/ContextImpl.java | 211 +++++++++--------- .../windowing/WindowFunctionExecutor.java | 2 +- .../windowing/WindowFunctionExecutorTest.java | 8 + 4 files changed, 117 insertions(+), 126 deletions(-) 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 85472003def00..013b5f2da6e01 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 @@ -250,28 +250,6 @@ public interface Context { */ 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 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 - * @deprecated in favor of using {@link #newOutputMessage(String, Schema)} - */ - CompletableFuture publish(String topicName, O object, String schemaOrSerdeClassName, Map messageConf); - /** * New output message using schema for serializing to the topic * 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 cb380cca6e65b..f0d5ade6cdf76 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 @@ -329,118 +329,22 @@ public CompletableFuture publish(String topicName, O object) { @SuppressWarnings("unchecked") @Override public CompletableFuture publish(String topicName, O object, String schemaOrSerdeClassName) { - return publish(topicName, object, schemaOrSerdeClassName, (Map) null); + return publish(topicName, object, (Schema) topicSchema.getSchema(topicName, object, schemaOrSerdeClassName, false)); } @SuppressWarnings("unchecked") - @Override - public CompletableFuture publish(String topicName, O object, String schemaOrSerdeClassName, Map messageConf) { - return publish(topicName, object, (Schema) topicSchema.getSchema(topicName, object, schemaOrSerdeClassName, false), messageConf); - } - @Override public TypedMessageBuilder newOutputMessage(String topicName, Schema schema) throws PulsarClientException { - final TypedMessageBuilder underlyingBuilder = getProducer(topicName, schema).newMessage(); - return new TypedMessageBuilder() { - - @Override - public MessageId send() throws PulsarClientException { - try { - return sendAsync().get(); - } catch (ExecutionException e) { - Throwable t = e.getCause(); - if (t instanceof PulsarClientException) { - throw (PulsarClientException) t; - } else { - throw new PulsarClientException(t); - } - } catch (InterruptedException e) { - Thread.currentThread().interrupt(); - throw new PulsarClientException(e); - } - } - - @Override - public CompletableFuture sendAsync() { - return underlyingBuilder.sendAsync() - .whenComplete((result, cause) -> { - if (null != cause) { - statsManager.incrSysExceptions(cause); - logger.error("Failed to publish to topic {} with error {}", topicName, cause); - } - }); - } - - @Override - public TypedMessageBuilder key(String key) { - underlyingBuilder.key(key); - return this; - } - - @Override - public TypedMessageBuilder keyBytes(byte[] key) { - underlyingBuilder.keyBytes(key); - return this; - } - - @Override - public TypedMessageBuilder value(O value) { - underlyingBuilder.value(value); - return this; - } - - @Override - public TypedMessageBuilder property(String name, String value) { - underlyingBuilder.property(name, value); - return this; - } - - @Override - public TypedMessageBuilder properties(Map properties) { - underlyingBuilder.properties(properties); - return this; - } - - @Override - public TypedMessageBuilder eventTime(long timestamp) { - underlyingBuilder.eventTime(timestamp); - return this; - } - - @Override - public TypedMessageBuilder sequenceId(long sequenceId) { - underlyingBuilder.sequenceId(sequenceId); - return this; - } - - @Override - public TypedMessageBuilder replicationClusters(List clusters) { - underlyingBuilder.replicationClusters(clusters); - return this; - } - - @Override - public TypedMessageBuilder disableReplication() { - underlyingBuilder.disableReplication(); - return this; - } - - @Override - public TypedMessageBuilder loadConf(Map config) { - underlyingBuilder.loadConf(config); - return this; - } - }; + MessageBuilderImpl messageBuilder = new MessageBuilderImpl<>(); + TypedMessageBuilder typedMessageBuilder = getProducer(topicName, schema).newMessage(); + messageBuilder.setUnderlyingBuilder(typedMessageBuilder); + return messageBuilder; } @SuppressWarnings("unchecked") - public CompletableFuture publish(String topicName, O object, Schema schema, Map messageConf) { + public CompletableFuture publish(String topicName, O object, Schema schema) { try { - TypedMessageBuilder messageBuilder = newOutputMessage(topicName, schema); - if (messageConf != null) { - messageBuilder.loadConf(messageConf); - } - return messageBuilder.value(object).sendAsync().thenApply(msgId -> null); + return newOutputMessage(topicName, schema).value(object).sendAsync().thenApply(msgId -> null); } catch (PulsarClientException e) { logger.error("Failed to create Producer while doing user publish", e); return FutureUtil.failedFuture(e); @@ -528,4 +432,105 @@ public Map getMetrics() { } return metricsMap; } + + class MessageBuilderImpl implements TypedMessageBuilder { + private TypedMessageBuilder underlyingBuilder; + @Override + public MessageId send() throws PulsarClientException { + try { + return sendAsync().get(); + } catch (ExecutionException e) { + Throwable t = e.getCause(); + if (t instanceof PulsarClientException) { + throw (PulsarClientException) t; + } else { + throw new PulsarClientException(t); + } + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new PulsarClientException(e); + } + } + + @Override + public CompletableFuture sendAsync() { + return underlyingBuilder.sendAsync() + .whenComplete((result, cause) -> { + if (null != cause) { + statsManager.incrSysExceptions(cause); + logger.error("Failed to publish to topic with error {}", cause); + } + }); + } + + @Override + public TypedMessageBuilder key(String key) { + underlyingBuilder.key(key); + return this; + } + + @Override + public TypedMessageBuilder keyBytes(byte[] key) { + underlyingBuilder.keyBytes(key); + return this; + } + + @Override + public TypedMessageBuilder orderingKey(byte[] orderingKey) { + underlyingBuilder.orderingKey(orderingKey); + return this; + } + + @Override + public TypedMessageBuilder value(O value) { + underlyingBuilder.value(value); + return this; + } + + @Override + public TypedMessageBuilder property(String name, String value) { + underlyingBuilder.property(name, value); + return this; + } + + @Override + public TypedMessageBuilder properties(Map properties) { + underlyingBuilder.properties(properties); + return this; + } + + @Override + public TypedMessageBuilder eventTime(long timestamp) { + underlyingBuilder.eventTime(timestamp); + return this; + } + + @Override + public TypedMessageBuilder sequenceId(long sequenceId) { + underlyingBuilder.sequenceId(sequenceId); + return this; + } + + @Override + public TypedMessageBuilder replicationClusters(List clusters) { + underlyingBuilder.replicationClusters(clusters); + return this; + } + + @Override + public TypedMessageBuilder disableReplication() { + underlyingBuilder.disableReplication(); + return this; + } + + @Override + public TypedMessageBuilder loadConf(Map config) { + underlyingBuilder.loadConf(config); + return this; + } + + public void setUnderlyingBuilder(TypedMessageBuilder underlyingBuilder) { + this.underlyingBuilder = underlyingBuilder; + } + } } \ No newline at end of file diff --git a/pulsar-functions/windowing/src/main/java/org/apache/pulsar/functions/windowing/WindowFunctionExecutor.java b/pulsar-functions/windowing/src/main/java/org/apache/pulsar/functions/windowing/WindowFunctionExecutor.java index 19459499cd64e..0fcc33e59c0bb 100644 --- a/pulsar-functions/windowing/src/main/java/org/apache/pulsar/functions/windowing/WindowFunctionExecutor.java +++ b/pulsar-functions/windowing/src/main/java/org/apache/pulsar/functions/windowing/WindowFunctionExecutor.java @@ -282,7 +282,7 @@ public O process(I input, Context context) throws Exception { this.windowManager.add(record, ts, record); } else { if (this.windowConfig.getLateDataTopic() != null) { - context.publish(this.windowConfig.getLateDataTopic(), input); + context.newOutputMessage(this.windowConfig.getLateDataTopic(), null).value(input).sendAsync(); } else { log.info(String.format( "Received a late tuple %s with ts %d. This will not be " + "processed" diff --git a/pulsar-functions/windowing/src/test/java/org/apache/pulsar/functions/windowing/WindowFunctionExecutorTest.java b/pulsar-functions/windowing/src/test/java/org/apache/pulsar/functions/windowing/WindowFunctionExecutorTest.java index 88ecebe53650f..ac96b44a65f62 100644 --- a/pulsar-functions/windowing/src/test/java/org/apache/pulsar/functions/windowing/WindowFunctionExecutorTest.java +++ b/pulsar-functions/windowing/src/test/java/org/apache/pulsar/functions/windowing/WindowFunctionExecutorTest.java @@ -20,6 +20,7 @@ import com.google.gson.Gson; import lombok.extern.slf4j.Slf4j; +import org.apache.pulsar.client.api.TypedMessageBuilder; import org.apache.pulsar.functions.api.Context; import org.apache.pulsar.functions.api.Record; @@ -36,8 +37,11 @@ import java.util.List; import java.util.Map; import java.util.Optional; +import java.util.concurrent.CompletableFuture; import java.util.function.Function; +import static org.mockito.Matchers.anyObject; +import static org.mockito.Matchers.anyString; import static org.testng.Assert.assertEquals; import static org.testng.Assert.fail; import static org.testng.internal.junit.ArrayAsserts.assertArrayEquals; @@ -204,6 +208,10 @@ public void testExecuteWithLateTupleStream() throws Exception { windowConfig.setLateDataTopic("$late"); Mockito.doReturn(Optional.of(new Gson().fromJson(new Gson().toJson(windowConfig), Map.class))) .when(context).getUserConfigValue(WindowConfig.WINDOW_CONFIG_KEY); + TypedMessageBuilder typedMessageBuilder = Mockito.mock(TypedMessageBuilder.class); + Mockito.when(typedMessageBuilder.value(anyString())).thenReturn(typedMessageBuilder); + Mockito.when(typedMessageBuilder.sendAsync()).thenReturn(CompletableFuture.anyOf()); + Mockito.when(context.newOutputMessage(anyString(), anyObject())).thenReturn(typedMessageBuilder); long[] timestamps = {603, 605, 607, 618, 626, 636, 600}; List events = new ArrayList<>(timestamps.length); From f046fc1d268a27b4dd995b488c5b82f7ba9b4a1d Mon Sep 17 00:00:00 2001 From: "xiaolong.ran" Date: Wed, 24 Apr 2019 11:00:31 +0800 Subject: [PATCH 07/16] fix function example Signed-off-by: xiaolong.ran --- pulsar-functions/api-java/pom.xml | 2 +- .../pulsar/functions/instance/ContextImplTest.java | 2 +- .../pulsar/functions/api/examples/PublishFunction.java | 4 ++-- .../api/examples/PublishFunctionWithMessageConf.java | 5 +++-- .../functions/api/examples/UserPublishFunction.java | 9 +++++++-- 5 files changed, 14 insertions(+), 8 deletions(-) diff --git a/pulsar-functions/api-java/pom.xml b/pulsar-functions/api-java/pom.xml index fe6554b708bf9..33e2ea05c0a22 100644 --- a/pulsar-functions/api-java/pom.xml +++ b/pulsar-functions/api-java/pom.xml @@ -45,7 +45,7 @@ org.apache.pulsar pulsar-client-api - 2.4.0-SNAPSHOT + ${project.version} compile 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 9d2579c78b6af..7eb7aae5305f6 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 @@ -153,6 +153,6 @@ public void testGetStateStateEnabled() throws Exception { @Test public void testPublishUsingDefaultSchema() throws Exception { - context.publish("sometopic", "Somevalue"); + context.newOutputMessage("sometopic", null).value("Somevalue").sendAsync(); } } diff --git a/pulsar-functions/java-examples/src/main/java/org/apache/pulsar/functions/api/examples/PublishFunction.java b/pulsar-functions/java-examples/src/main/java/org/apache/pulsar/functions/api/examples/PublishFunction.java index 0c7cb56ea825c..87853054e5625 100644 --- a/pulsar-functions/java-examples/src/main/java/org/apache/pulsar/functions/api/examples/PublishFunction.java +++ b/pulsar-functions/java-examples/src/main/java/org/apache/pulsar/functions/api/examples/PublishFunction.java @@ -19,6 +19,7 @@ package org.apache.pulsar.functions.api.examples; 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.functions.api.Context; import org.apache.pulsar.functions.api.Function; @@ -33,8 +34,7 @@ public Void process(String input, Context context) { String publishTopic = (String) context.getUserConfigValueOrDefault("publish-topic", "publishtopic"); String output = String.format("%s!", input); try { - TypedMessageBuilder msgBuilder = context.newOutputMessage(publishTopic, null); - msgBuilder.value(output).sendAsync(); + context.newOutputMessage(publishTopic, Schema.STRING).value(output).sendAsync(); } catch (PulsarClientException e) { e.printStackTrace(); } 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 bb034b24f3383..2ea20f93f3954 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 @@ -19,6 +19,7 @@ package org.apache.pulsar.functions.api.examples; 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.functions.api.Context; import org.apache.pulsar.functions.api.Function; @@ -42,8 +43,8 @@ public Void process(String input, Context context) { properties.putAll(context.getCurrentRecord().getProperties()); try { - TypedMessageBuilder msgBuilder = context.newOutputMessage(publishTopic, null); - msgBuilder.value(output).properties(properties). + context.newOutputMessage(publishTopic, Schema.STRING) + .value(output).properties(properties). key(context.getCurrentRecord().getKey().get()). eventTime(System.currentTimeMillis()).sendAsync(); } catch (PulsarClientException e) { diff --git a/pulsar-functions/java-examples/src/main/java/org/apache/pulsar/functions/api/examples/UserPublishFunction.java b/pulsar-functions/java-examples/src/main/java/org/apache/pulsar/functions/api/examples/UserPublishFunction.java index ec3fd95ca80b3..e597c89c256b7 100644 --- a/pulsar-functions/java-examples/src/main/java/org/apache/pulsar/functions/api/examples/UserPublishFunction.java +++ b/pulsar-functions/java-examples/src/main/java/org/apache/pulsar/functions/api/examples/UserPublishFunction.java @@ -20,6 +20,7 @@ import java.util.Optional; +import org.apache.pulsar.client.api.PulsarClientException; import org.apache.pulsar.functions.api.Context; import org.apache.pulsar.functions.api.Function; @@ -31,8 +32,12 @@ public class UserPublishFunction implements Function { @Override public Void process(String input, Context context) { Optional topicToWrite = context.getUserConfigValue("topic"); - if (topicToWrite.get() != null) { - context.publish((String) topicToWrite.get(), input); + if (topicToWrite.isPresent()) { + try { + context.newOutputMessage((String) topicToWrite.get(),null).value(input).sendAsync(); + } catch (PulsarClientException e) { + e.printStackTrace(); + } } return null; } From 94ab85565c412274109dc926e0e23c8856858594 Mon Sep 17 00:00:00 2001 From: "xiaolong.ran" Date: Wed, 24 Apr 2019 11:01:49 +0800 Subject: [PATCH 08/16] fix a little Signed-off-by: xiaolong.ran --- .../pulsar/functions/api/examples/UserPublishFunction.java | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/pulsar-functions/java-examples/src/main/java/org/apache/pulsar/functions/api/examples/UserPublishFunction.java b/pulsar-functions/java-examples/src/main/java/org/apache/pulsar/functions/api/examples/UserPublishFunction.java index e597c89c256b7..41c4d7dce0fe4 100644 --- a/pulsar-functions/java-examples/src/main/java/org/apache/pulsar/functions/api/examples/UserPublishFunction.java +++ b/pulsar-functions/java-examples/src/main/java/org/apache/pulsar/functions/api/examples/UserPublishFunction.java @@ -21,6 +21,7 @@ import java.util.Optional; import org.apache.pulsar.client.api.PulsarClientException; +import org.apache.pulsar.client.api.Schema; import org.apache.pulsar.functions.api.Context; import org.apache.pulsar.functions.api.Function; @@ -34,7 +35,7 @@ public Void process(String input, Context context) { Optional topicToWrite = context.getUserConfigValue("topic"); if (topicToWrite.isPresent()) { try { - context.newOutputMessage((String) topicToWrite.get(),null).value(input).sendAsync(); + context.newOutputMessage((String) topicToWrite.get(), Schema.STRING).value(input).sendAsync(); } catch (PulsarClientException e) { e.printStackTrace(); } From 656b921f5a893a1e102699e958d1a5cd7f2b7861 Mon Sep 17 00:00:00 2001 From: "xiaolong.ran" Date: Tue, 30 Apr 2019 11:38:25 +0800 Subject: [PATCH 09/16] fix integration tests Signed-off-by: xiaolong.ran --- .../pulsar/functions/windowing/WindowFunctionExecutor.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pulsar-functions/windowing/src/main/java/org/apache/pulsar/functions/windowing/WindowFunctionExecutor.java b/pulsar-functions/windowing/src/main/java/org/apache/pulsar/functions/windowing/WindowFunctionExecutor.java index 0fcc33e59c0bb..19459499cd64e 100644 --- a/pulsar-functions/windowing/src/main/java/org/apache/pulsar/functions/windowing/WindowFunctionExecutor.java +++ b/pulsar-functions/windowing/src/main/java/org/apache/pulsar/functions/windowing/WindowFunctionExecutor.java @@ -282,7 +282,7 @@ public O process(I input, Context context) throws Exception { this.windowManager.add(record, ts, record); } else { if (this.windowConfig.getLateDataTopic() != null) { - context.newOutputMessage(this.windowConfig.getLateDataTopic(), null).value(input).sendAsync(); + context.publish(this.windowConfig.getLateDataTopic(), input); } else { log.info(String.format( "Received a late tuple %s with ts %d. This will not be " + "processed" From 037c28aeb72f866e9e2771492ac0e95c065dae20 Mon Sep 17 00:00:00 2001 From: "xiaolong.ran" Date: Tue, 30 Apr 2019 17:49:59 +0800 Subject: [PATCH 10/16] fix PublishFunctionWithMessageConf Signed-off-by: xiaolong.ran --- .../api/examples/PublishFunctionWithMessageConf.java | 10 ++++++---- .../functions/windowing/WindowFunctionExecutor.java | 2 +- 2 files changed, 7 insertions(+), 5 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 2ea20f93f3954..8b3dae807c17e 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 @@ -43,10 +43,12 @@ public Void process(String input, Context context) { properties.putAll(context.getCurrentRecord().getProperties()); try { - context.newOutputMessage(publishTopic, Schema.STRING) - .value(output).properties(properties). - key(context.getCurrentRecord().getKey().get()). - eventTime(System.currentTimeMillis()).sendAsync(); + TypedMessageBuilder messageBuilder = context.newOutputMessage(publishTopic, Schema.STRING). + value(output).properties(properties); + if (context.getCurrentRecord().getKey().isPresent()){ + messageBuilder.key(context.getCurrentRecord().getKey().get()); + } + messageBuilder.eventTime(System.currentTimeMillis()).sendAsync(); } catch (PulsarClientException e) { e.printStackTrace(); } diff --git a/pulsar-functions/windowing/src/main/java/org/apache/pulsar/functions/windowing/WindowFunctionExecutor.java b/pulsar-functions/windowing/src/main/java/org/apache/pulsar/functions/windowing/WindowFunctionExecutor.java index 19459499cd64e..0fcc33e59c0bb 100644 --- a/pulsar-functions/windowing/src/main/java/org/apache/pulsar/functions/windowing/WindowFunctionExecutor.java +++ b/pulsar-functions/windowing/src/main/java/org/apache/pulsar/functions/windowing/WindowFunctionExecutor.java @@ -282,7 +282,7 @@ public O process(I input, Context context) throws Exception { this.windowManager.add(record, ts, record); } else { if (this.windowConfig.getLateDataTopic() != null) { - context.publish(this.windowConfig.getLateDataTopic(), input); + context.newOutputMessage(this.windowConfig.getLateDataTopic(), null).value(input).sendAsync(); } else { log.info(String.format( "Received a late tuple %s with ts %d. This will not be " + "processed" From 36e905b6ae12f405cabe6b3dfe6ac077716ee9a4 Mon Sep 17 00:00:00 2001 From: "xiaolong.ran" Date: Fri, 3 May 2019 01:57:37 +0800 Subject: [PATCH 11/16] fix error info Signed-off-by: xiaolong.ran --- .../apache/pulsar/functions/api/examples/PublishFunction.java | 2 +- .../functions/api/examples/PublishFunctionWithMessageConf.java | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/pulsar-functions/java-examples/src/main/java/org/apache/pulsar/functions/api/examples/PublishFunction.java b/pulsar-functions/java-examples/src/main/java/org/apache/pulsar/functions/api/examples/PublishFunction.java index 87853054e5625..149cce60cef0f 100644 --- a/pulsar-functions/java-examples/src/main/java/org/apache/pulsar/functions/api/examples/PublishFunction.java +++ b/pulsar-functions/java-examples/src/main/java/org/apache/pulsar/functions/api/examples/PublishFunction.java @@ -36,7 +36,7 @@ public Void process(String input, Context context) { try { context.newOutputMessage(publishTopic, Schema.STRING).value(output).sendAsync(); } catch (PulsarClientException e) { - e.printStackTrace(); + context.getLogger().error(e.toString()); } return null; } 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 8b3dae807c17e..8435a265c0c34 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 @@ -50,7 +50,7 @@ public Void process(String input, Context context) { } messageBuilder.eventTime(System.currentTimeMillis()).sendAsync(); } catch (PulsarClientException e) { - e.printStackTrace(); + context.getLogger().error(e.toString()); } return null; } From d4759856d0a84b91dc8976e5d154a467b4073f17 Mon Sep 17 00:00:00 2001 From: "xiaolong.ran" Date: Sat, 4 May 2019 01:22:37 +0800 Subject: [PATCH 12/16] use typedMessageBuilderPublish instead of PublishFunctionWithMessageConf Signed-off-by: xiaolong.ran --- .../pulsar/functions/worker/PulsarFunctionPublishTest.java | 2 +- ...onWithMessageConf.java => typedMessageBuilderPublish.java} | 2 +- .../tests/integration/functions/PulsarFunctionsTestBase.java | 4 ++-- 3 files changed, 4 insertions(+), 4 deletions(-) rename pulsar-functions/java-examples/src/main/java/org/apache/pulsar/functions/api/examples/{PublishFunctionWithMessageConf.java => typedMessageBuilderPublish.java} (96%) 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 index ef88fddc7e156..625afc3464829 100644 --- 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 @@ -279,7 +279,7 @@ protected static FunctionConfig createFunctionConfig(String tenant, String names functionConfig.setSubName(subscriptionName); functionConfig.setInputs(Collections.singleton(sourceTopic)); functionConfig.setAutoAck(true); - functionConfig.setClassName("org.apache.pulsar.functions.api.examples.PublishFunctionWithMessageConf"); + functionConfig.setClassName("org.apache.pulsar.functions.api.examples.typedMessageBuilderPublish"); functionConfig.setRuntime(FunctionConfig.Runtime.JAVA); Map userConfig = new HashMap<>(); userConfig.put("publish-topic", publishTopic); 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/typedMessageBuilderPublish.java similarity index 96% rename from pulsar-functions/java-examples/src/main/java/org/apache/pulsar/functions/api/examples/PublishFunctionWithMessageConf.java rename to pulsar-functions/java-examples/src/main/java/org/apache/pulsar/functions/api/examples/typedMessageBuilderPublish.java index 8435a265c0c34..92ed913a5eb8a 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/typedMessageBuilderPublish.java @@ -32,7 +32,7 @@ * to publish to a desired topic based on config and setting various message configurations to be passed along. * */ -public class PublishFunctionWithMessageConf implements Function { +public class typedMessageBuilderPublish implements Function { @Override public Void process(String input, Context context) { String publishTopic = (String) context.getUserConfigValueOrDefault("publish-topic", "publishtopic"); 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 e7173896868e9..708876e7dae3e 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 @@ -74,7 +74,7 @@ public void teardownFunctionWorkers() { "org.apache.pulsar.functions.api.examples.ExclamationFunction"; public static final String PUBLISH_JAVA_CLASS = - "org.apache.pulsar.functions.api.examples.PublishFunctionWithMessageConf"; + "org.apache.pulsar.functions.api.examples.typedMessageBuilderPublish"; public static final String EXCLAMATION_PYTHON_CLASS = @@ -86,7 +86,7 @@ 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 PUBLISH_PYTHON_CLASS = "publish_function_with_message_conf.typedMessageBuilderPublish"; public static final String EXCLAMATION_PYTHON_FILE = "exclamation_function.py"; public static final String EXCLAMATION_WITH_DEPS_PYTHON_FILE = "exclamation_with_extra_deps.py"; From 9c83892490f5d3868713ff84f47d59c3a670f6a9 Mon Sep 17 00:00:00 2001 From: "xiaolong.ran" Date: Sat, 4 May 2019 02:27:44 +0800 Subject: [PATCH 13/16] fix a little Signed-off-by: xiaolong.ran --- .../pulsar/functions/worker/PulsarFunctionPublishTest.java | 2 +- ...ageBuilderPublish.java => TypedMessageBuilderPublish.java} | 2 +- .../tests/integration/functions/PulsarFunctionsTestBase.java | 4 ++-- 3 files changed, 4 insertions(+), 4 deletions(-) rename pulsar-functions/java-examples/src/main/java/org/apache/pulsar/functions/api/examples/{typedMessageBuilderPublish.java => TypedMessageBuilderPublish.java} (97%) 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 index 625afc3464829..e3b41313188b6 100644 --- 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 @@ -279,7 +279,7 @@ protected static FunctionConfig createFunctionConfig(String tenant, String names functionConfig.setSubName(subscriptionName); functionConfig.setInputs(Collections.singleton(sourceTopic)); functionConfig.setAutoAck(true); - functionConfig.setClassName("org.apache.pulsar.functions.api.examples.typedMessageBuilderPublish"); + functionConfig.setClassName("org.apache.pulsar.functions.api.examples.TypedMessageBuilderPublish"); functionConfig.setRuntime(FunctionConfig.Runtime.JAVA); Map userConfig = new HashMap<>(); userConfig.put("publish-topic", publishTopic); diff --git a/pulsar-functions/java-examples/src/main/java/org/apache/pulsar/functions/api/examples/typedMessageBuilderPublish.java b/pulsar-functions/java-examples/src/main/java/org/apache/pulsar/functions/api/examples/TypedMessageBuilderPublish.java similarity index 97% rename from pulsar-functions/java-examples/src/main/java/org/apache/pulsar/functions/api/examples/typedMessageBuilderPublish.java rename to pulsar-functions/java-examples/src/main/java/org/apache/pulsar/functions/api/examples/TypedMessageBuilderPublish.java index 92ed913a5eb8a..327d37187b980 100644 --- a/pulsar-functions/java-examples/src/main/java/org/apache/pulsar/functions/api/examples/typedMessageBuilderPublish.java +++ b/pulsar-functions/java-examples/src/main/java/org/apache/pulsar/functions/api/examples/TypedMessageBuilderPublish.java @@ -32,7 +32,7 @@ * to publish to a desired topic based on config and setting various message configurations to be passed along. * */ -public class typedMessageBuilderPublish implements Function { +public class TypedMessageBuilderPublish implements Function { @Override public Void process(String input, Context context) { String publishTopic = (String) context.getUserConfigValueOrDefault("publish-topic", "publishtopic"); 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 708876e7dae3e..fb6928c39711b 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 @@ -74,7 +74,7 @@ public void teardownFunctionWorkers() { "org.apache.pulsar.functions.api.examples.ExclamationFunction"; public static final String PUBLISH_JAVA_CLASS = - "org.apache.pulsar.functions.api.examples.typedMessageBuilderPublish"; + "org.apache.pulsar.functions.api.examples.TypedMessageBuilderPublish"; public static final String EXCLAMATION_PYTHON_CLASS = @@ -86,7 +86,7 @@ public void teardownFunctionWorkers() { public static final String EXCLAMATION_PYTHONZIP_CLASS = "exclamation"; - public static final String PUBLISH_PYTHON_CLASS = "publish_function_with_message_conf.typedMessageBuilderPublish"; + public static final String PUBLISH_PYTHON_CLASS = "publish_function_with_message_conf.TypedMessageBuilderPublish"; public static final String EXCLAMATION_PYTHON_FILE = "exclamation_function.py"; public static final String EXCLAMATION_WITH_DEPS_PYTHON_FILE = "exclamation_with_extra_deps.py"; From 6280e561386ba1c2fe0f6a87943ecbd40588d96e Mon Sep 17 00:00:00 2001 From: "xiaolong.ran" Date: Sat, 4 May 2019 13:19:29 +0800 Subject: [PATCH 14/16] fix integration tests Signed-off-by: xiaolong.ran --- ...on_with_message_conf.py => typed_message_builder_publish.py} | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) rename pulsar-functions/python-examples/{publish_function_with_message_conf.py => typed_message_builder_publish.py} (97%) diff --git a/pulsar-functions/python-examples/publish_function_with_message_conf.py b/pulsar-functions/python-examples/typed_message_builder_publish.py similarity index 97% rename from pulsar-functions/python-examples/publish_function_with_message_conf.py rename to pulsar-functions/python-examples/typed_message_builder_publish.py index 79aac0239fa55..c6697a716d071 100644 --- a/pulsar-functions/python-examples/publish_function_with_message_conf.py +++ b/pulsar-functions/python-examples/typed_message_builder_publish.py @@ -23,7 +23,7 @@ # Example function that uses the built in publish function in the context # to publish to a desired topic based on config -class PublishFunctionWithMessageConf(Function): +class TypedMessageBuilderPublish(Function): def __init__(self): pass From 3c2e8b252241df49480fc9e7bab87e7c86b7ae0d Mon Sep 17 00:00:00 2001 From: "xiaolong.ran" Date: Sat, 4 May 2019 13:24:38 +0800 Subject: [PATCH 15/16] fix java unit test Signed-off-by: xiaolong.ran --- .../tests/integration/functions/PulsarFunctionsTestBase.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) 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 fb6928c39711b..788a1c182c704 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 @@ -86,7 +86,7 @@ public void teardownFunctionWorkers() { public static final String EXCLAMATION_PYTHONZIP_CLASS = "exclamation"; - public static final String PUBLISH_PYTHON_CLASS = "publish_function_with_message_conf.TypedMessageBuilderPublish"; + public static final String PUBLISH_PYTHON_CLASS = "typed_message_builder_publish.TypedMessageBuilderPublish"; public static final String EXCLAMATION_PYTHON_FILE = "exclamation_function.py"; public static final String EXCLAMATION_WITH_DEPS_PYTHON_FILE = "exclamation_with_extra_deps.py"; From 80d3ae46b6374058dfd8b17ea2cda6ea8384b6d2 Mon Sep 17 00:00:00 2001 From: "xiaolong.ran" Date: Sat, 4 May 2019 14:53:23 +0800 Subject: [PATCH 16/16] fix integration test Signed-off-by: xiaolong.ran --- .../tests/integration/functions/PulsarFunctionsTestBase.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) 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 788a1c182c704..ad0ed2c7b65b5 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 @@ -91,7 +91,7 @@ public void teardownFunctionWorkers() { 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"; + public static final String PUBLISH_FUNCTION_PYTHON_FILE = "typed_message_builder_publish.py"; protected static String getExclamationClass(Runtime runtime,