You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by GitBox <gi...@apache.org> on 2018/09/19 17:01:23 UTC
[GitHub] srkukarni closed pull request #2606: Add support for dead letter
topics for java functions
srkukarni closed pull request #2606: Add support for dead letter topics for java functions
URL: https://github.com/apache/incubator-pulsar/pull/2606
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git a/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdFunctions.java b/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdFunctions.java
index e8c874059c..5284b59f5b 100644
--- a/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdFunctions.java
+++ b/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdFunctions.java
@@ -24,7 +24,7 @@
import static org.apache.bookkeeper.common.concurrent.FutureUtils.result;
import static org.apache.commons.lang.StringUtils.isBlank;
import static org.apache.commons.lang.StringUtils.isNotBlank;
-import static org.apache.commons.lang3.StringUtils.isNotBlank;
+import static org.apache.commons.lang.StringUtils.isNotEmpty;
import static org.apache.pulsar.common.naming.TopicName.DEFAULT_NAMESPACE;
import static org.apache.pulsar.common.naming.TopicName.PUBLIC_TENANT;
import static org.apache.pulsar.functions.utils.Utils.fileExists;
@@ -45,7 +45,6 @@
import io.netty.buffer.Unpooled;
import java.io.File;
-import java.io.FileOutputStream;
import java.io.IOException;
import java.lang.reflect.Type;
import java.net.MalformedURLException;
@@ -79,12 +78,12 @@
import org.apache.pulsar.functions.proto.Function.ConsumerSpec;
import org.apache.pulsar.functions.proto.Function.FunctionDetails;
import org.apache.pulsar.functions.proto.Function.Resources;
+import org.apache.pulsar.functions.proto.Function.RetryDetails;
import org.apache.pulsar.functions.proto.Function.SinkSpec;
import org.apache.pulsar.functions.proto.Function.SourceSpec;
import org.apache.pulsar.functions.proto.Function.SubscriptionType;
import org.apache.pulsar.functions.runtime.ProcessRuntimeFactory;
import org.apache.pulsar.functions.runtime.RuntimeSpawner;
-import org.apache.pulsar.functions.utils.ConsumerConfig;
import org.apache.pulsar.functions.utils.FunctionConfig;
import org.apache.pulsar.functions.utils.Reflections;
import org.apache.pulsar.functions.utils.Utils;
@@ -321,6 +320,10 @@ void processArguments() throws Exception {
protected Long DEPRECATED_timeoutMs;
@Parameter(names = "--timeout-ms", description = "The message timeout in milliseconds")
protected Long timeoutMs;
+ @Parameter(names = "--max-message-retries", description = "How many times should we try to process a message before giving up")
+ protected Integer maxMessageRetries = -1;
+ @Parameter(names = "--dead-letter-topic", description = "The topic where all messages which could not be processed successfully are sent")
+ protected String deadLetterTopic;
protected FunctionConfig functionConfig;
protected String userCodeFile;
@@ -464,6 +467,13 @@ void processArguments() throws Exception {
functionConfig.setAutoAck(autoAck);
+ if (null != maxMessageRetries) {
+ functionConfig.setMaxMessageRetries(maxMessageRetries);
+ }
+ if (null != deadLetterTopic) {
+ functionConfig.setDeadLetterTopic(deadLetterTopic);
+ }
+
if (null != jarFile) {
functionConfig.setJar(jarFile);
}
@@ -717,6 +727,15 @@ protected FunctionDetails convert(FunctionConfig functionConfig)
Utils.convertProcessingGuarantee(functionConfig.getProcessingGuarantees()));
}
+ if (functionConfig.getMaxMessageRetries() >= 0) {
+ RetryDetails.Builder retryBuilder = RetryDetails.newBuilder();
+ retryBuilder.setMaxMessageRetries(functionConfig.getMaxMessageRetries());
+ if (isNotEmpty(functionConfig.getDeadLetterTopic())) {
+ retryBuilder.setDeadLetterTopic(functionConfig.getDeadLetterTopic());
+ }
+ functionDetailsBuilder.setRetryDetails(retryBuilder);
+ }
+
Map<String, Object> configs = new HashMap<>();
configs.putAll(functionConfig.getUserConfig());
diff --git a/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/JavaInstanceRunnable.java b/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/JavaInstanceRunnable.java
index b3f86eae0e..1e07516f59 100644
--- a/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/JavaInstanceRunnable.java
+++ b/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/JavaInstanceRunnable.java
@@ -547,6 +547,11 @@ public void setupInput(ContextImpl contextImpl) throws Exception {
if (sourceSpec.getTimeoutMs() > 0 ) {
pulsarSourceConfig.setTimeoutMs(sourceSpec.getTimeoutMs());
}
+
+ if (this.instanceConfig.getFunctionDetails().getRetryDetails() != null) {
+ pulsarSourceConfig.setMaxMessageRetries(this.instanceConfig.getFunctionDetails().getRetryDetails().getMaxMessageRetries());
+ pulsarSourceConfig.setDeadLetterTopic(this.instanceConfig.getFunctionDetails().getRetryDetails().getDeadLetterTopic());
+ }
object = new PulsarSource(this.client, pulsarSourceConfig,
FunctionDetailsUtils.getFullyQualifiedName(this.instanceConfig.getFunctionDetails()));
} else {
diff --git a/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/source/PulsarSource.java b/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/source/PulsarSource.java
index 6eed8e0142..afac7821cd 100644
--- a/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/source/PulsarSource.java
+++ b/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/source/PulsarSource.java
@@ -31,14 +31,7 @@
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
-import org.apache.pulsar.client.api.Consumer;
-import org.apache.pulsar.client.api.ConsumerBuilder;
-import org.apache.pulsar.client.api.ConsumerCryptoFailureAction;
-import org.apache.pulsar.client.api.Message;
-import org.apache.pulsar.client.api.MessageListener;
-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.*;
import org.apache.pulsar.client.impl.MultiTopicsConsumerImpl;
import org.apache.pulsar.client.impl.TopicMessageImpl;
import org.apache.pulsar.functions.api.Record;
@@ -97,6 +90,15 @@ public void open(Map<String, Object> config, SourceContext sourceContext) throws
cb.ackTimeout(pulsarSourceConfig.getTimeoutMs(), TimeUnit.MILLISECONDS);
}
+ if (pulsarSourceConfig.getMaxMessageRetries() >= 0) {
+ DeadLetterPolicy.DeadLetterPolicyBuilder deadLetterPolicyBuilder = DeadLetterPolicy.builder();
+ deadLetterPolicyBuilder.maxRedeliverCount(pulsarSourceConfig.getMaxMessageRetries());
+ if (pulsarSourceConfig.getDeadLetterTopic() != null && !pulsarSourceConfig.getDeadLetterTopic().isEmpty()) {
+ deadLetterPolicyBuilder.deadLetterTopic(pulsarSourceConfig.getDeadLetterTopic());
+ }
+ cb.deadLetterPolicy(deadLetterPolicyBuilder.build());
+ }
+
return cb.subscribeAsync();
}).collect(Collectors.toList()).stream().map(CompletableFuture::join).collect(Collectors.toList());
diff --git a/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/source/PulsarSourceConfig.java b/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/source/PulsarSourceConfig.java
index f1cb09b494..4e2afa7c24 100644
--- a/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/source/PulsarSourceConfig.java
+++ b/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/source/PulsarSourceConfig.java
@@ -37,6 +37,8 @@
private FunctionConfig.ProcessingGuarantees processingGuarantees;
SubscriptionType subscriptionType;
private String subscriptionName;
+ private int maxMessageRetries;
+ private String deadLetterTopic;
private Map<String, ConsumerConfig> topicSchema = new TreeMap<>();
diff --git a/pulsar-functions/java-examples/src/main/java/org/apache/pulsar/functions/api/examples/ConsistentlyFailingFunction.java b/pulsar-functions/java-examples/src/main/java/org/apache/pulsar/functions/api/examples/ConsistentlyFailingFunction.java
new file mode 100644
index 0000000000..792a5741ca
--- /dev/null
+++ b/pulsar-functions/java-examples/src/main/java/org/apache/pulsar/functions/api/examples/ConsistentlyFailingFunction.java
@@ -0,0 +1,38 @@
+/**
+ * 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;
+
+
+/**
+ * This Function simulates a pulsar function encountering failing on a particular message.
+ */
+public class ConsistentlyFailingFunction implements Function<String, String> {
+ @Override
+ public String process(String input, Context context) {
+ if (input.equals("FAIL")) {
+ throw new RuntimeException("Failed");
+ } else {
+ return "SUCCESS";
+ }
+ }
+}
+
diff --git a/pulsar-functions/proto/src/main/proto/Function.proto b/pulsar-functions/proto/src/main/proto/Function.proto
index a76cf8d5ed..482d901536 100644
--- a/pulsar-functions/proto/src/main/proto/Function.proto
+++ b/pulsar-functions/proto/src/main/proto/Function.proto
@@ -40,6 +40,11 @@ message Resources {
int64 disk = 3;
}
+message RetryDetails {
+ int32 maxMessageRetries = 1;
+ string deadLetterTopic = 2;
+}
+
message FunctionDetails {
enum Runtime {
JAVA = 0;
@@ -59,6 +64,7 @@ message FunctionDetails {
SinkSpec sink = 12;
Resources resources = 13;
string packageUrl = 14; //present only if function submitted with package-url
+ RetryDetails retryDetails = 15;
}
message ConsumerSpec {
diff --git a/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/FunctionConfig.java b/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/FunctionConfig.java
index 1335f8c067..dc368126f4 100644
--- a/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/FunctionConfig.java
+++ b/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/FunctionConfig.java
@@ -107,6 +107,8 @@
private Map<String, Object> userConfig;
private Runtime runtime;
private boolean autoAck;
+ private int maxMessageRetries;
+ private String deadLetterTopic;
private String subName;
@isPositiveNumber
private int parallelism;
diff --git a/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/validation/ValidatorImpls.java b/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/validation/ValidatorImpls.java
index f60f3c00f9..e8acc2839b 100644
--- a/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/validation/ValidatorImpls.java
+++ b/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/validation/ValidatorImpls.java
@@ -514,6 +514,10 @@ private static void doPythonChecks(FunctionConfig functionConfig, String name) {
if (functionConfig.getWindowConfig() != null) {
throw new IllegalArgumentException("There is currently no support windowing in python");
}
+
+ if (functionConfig.getMaxMessageRetries() >= 0) {
+ throw new IllegalArgumentException("Message retries not yet supported in python");
+ }
}
private static void verifyNoTopicClash(Collection<String> inputTopics, String outputTopic) throws IllegalArgumentException {
@@ -549,6 +553,14 @@ private static void doCommonChecks(FunctionConfig functionConfig) {
throw new IllegalArgumentException("Message timeout can only be specified with processing guarantee is "
+ FunctionConfig.ProcessingGuarantees.ATLEAST_ONCE.name());
}
+
+ if (functionConfig.getMaxMessageRetries() >= 0
+ && functionConfig.getProcessingGuarantees() == FunctionConfig.ProcessingGuarantees.EFFECTIVELY_ONCE) {
+ throw new IllegalArgumentException("MaxMessageRetries and Effectively once don't gel well");
+ }
+ if (functionConfig.getMaxMessageRetries() < 0 && !StringUtils.isEmpty(functionConfig.getDeadLetterTopic())) {
+ throw new IllegalArgumentException("Dead Letter Topic specified, however max retries is set to infinity");
+ }
}
private static Collection<String> collectAllInputTopics(FunctionConfig functionConfig) {
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services