You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by je...@apache.org on 2019/04/25 04:44:57 UTC
[pulsar] branch master updated: Function use negative acknowledge
(#4103)
This is an automated email from the ASF dual-hosted git repository.
jerrypeng pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/master by this push:
new 4e84d5b Function use negative acknowledge (#4103)
4e84d5b is described below
commit 4e84d5bdb9c74876150f67edb6eb05167f8ed943
Author: Boyang Jerry Peng <je...@gmail.com>
AuthorDate: Wed Apr 24 21:44:50 2019 -0700
Function use negative acknowledge (#4103)
* Functions use neg ack
* add neg ack for python functions
* adding tests
* refactoring maven project
* remove unnecessary edit
* fix pom due to maven bug
---
.../instance/stats/FunctionStatsManager.java | 3 -
.../functions/instance/stats/SinkStatsManager.java | 3 -
.../instance/stats/SourceStatsManager.java | 3 -
.../pulsar/functions/source/PulsarSource.java | 1 +
.../instance/src/main/python/function_stats.py | 8 +-
.../instance/src/main/python/python_instance.py | 6 +
tests/docker-images/java-test-functions/pom.xml | 81 ++++++++++
.../integration/functions/ExceptionFunction.java | 35 ++++
.../docker-images/latest-version-image/Dockerfile | 2 +
tests/docker-images/latest-version-image/pom.xml | 30 ++++
.../python-examples/exception_function.py | 29 ++++
tests/docker-images/pom.xml | 1 +
.../integration/functions/PulsarFunctionsTest.java | 176 +++++++++++++++++++++
.../functions/PulsarFunctionsTestBase.java | 7 +-
.../functions/utils/CommandGenerator.java | 2 +-
15 files changed, 371 insertions(+), 16 deletions(-)
diff --git a/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/stats/FunctionStatsManager.java b/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/stats/FunctionStatsManager.java
index b35c32a..7943d5a 100644
--- a/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/stats/FunctionStatsManager.java
+++ b/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/stats/FunctionStatsManager.java
@@ -437,8 +437,5 @@ public class FunctionStatsManager extends ComponentStatsManager{
statTotalRecordsReceived1min.clear();
_statTotalRecordsReceived1min = statTotalRecordsReceived1min.labels(metricsLabels);
-
- latestUserExceptions.clear();
- latestSystemExceptions.clear();
}
}
diff --git a/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/stats/SinkStatsManager.java b/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/stats/SinkStatsManager.java
index 1b2d8b0..3ed2aa1 100644
--- a/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/stats/SinkStatsManager.java
+++ b/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/stats/SinkStatsManager.java
@@ -191,9 +191,6 @@ public class SinkStatsManager extends ComponentStatsManager {
statTotalWritten1min.clear();
_statTotalWritten1min = statTotalWritten1min.labels(metricsLabels);
-
- latestSystemExceptions.clear();
- latestSinkExceptions.clear();
}
@Override
diff --git a/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/stats/SourceStatsManager.java b/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/stats/SourceStatsManager.java
index aac6720..f496709 100644
--- a/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/stats/SourceStatsManager.java
+++ b/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/stats/SourceStatsManager.java
@@ -190,9 +190,6 @@ public class SourceStatsManager extends ComponentStatsManager {
statTotalWritten1min.clear();
_statTotalWritten1min = statTotalWritten1min.labels(metricsLabels);
-
- latestSystemExceptions.clear();
- latestSourceExceptions.clear();
}
@Override
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 83e7566..fd81260 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
@@ -122,6 +122,7 @@ public class PulsarSource<T> extends PushSource<T> implements MessageListener<T>
if (pulsarSourceConfig.getProcessingGuarantees() == FunctionConfig.ProcessingGuarantees.EFFECTIVELY_ONCE) {
throw new RuntimeException("Failed to process message: " + message.getMessageId());
}
+ consumer.negativeAcknowledge(message);
})
.build();
diff --git a/pulsar-functions/instance/src/main/python/function_stats.py b/pulsar-functions/instance/src/main/python/function_stats.py
index 34793a5..63089b6 100644
--- a/pulsar-functions/instance/src/main/python/function_stats.py
+++ b/pulsar-functions/instance/src/main/python/function_stats.py
@@ -179,9 +179,9 @@ class Stats(object):
def add_user_exception(self, exception):
error = traceback.format_exc()
ts = int(time.time() * 1000) if sys.version_info.major >= 3 else long(time.time() * 1000)
- self.latest_sys_exception.append((error, ts))
- if len(self.latest_sys_exception) > 10:
- self.latest_sys_exception.pop(0)
+ self.latest_user_exception.append((error, ts))
+ if len(self.latest_user_exception) > 10:
+ self.latest_user_exception.pop(0)
# report exception via prometheus
try:
@@ -213,8 +213,6 @@ class Stats(object):
self.system_exceptions.labels(*exception_metric_labels).set(1.0)
def reset(self):
- self.latest_user_exception = []
- self.latest_sys_exception = []
self._stat_total_processed_successfully_1min._value.set(0.0)
self._stat_total_user_exceptions_1min._value.set(0.0)
self._stat_total_sys_exceptions_1min._value.set(0.0)
diff --git a/pulsar-functions/instance/src/main/python/python_instance.py b/pulsar-functions/instance/src/main/python/python_instance.py
index 8f740f1..86f8644 100644
--- a/pulsar-functions/instance/src/main/python/python_instance.py
+++ b/pulsar-functions/instance/src/main/python/python_instance.py
@@ -249,6 +249,8 @@ class PythonInstance(object):
except Exception as e:
Log.exception("Exception while executing user method")
self.stats.incr_total_user_exceptions(e)
+ # If function throws exception then send neg ack for input message back to broker
+ msg.consumer.negative_acknowledge(msg.message)
if self.log_topic_handler is not None:
log.remove_all_handlers()
@@ -260,6 +262,8 @@ class PythonInstance(object):
except Exception as e:
Log.error("Uncaught exception in Python instance: %s" % e);
self.stats.incr_total_sys_exceptions(e)
+ if msg:
+ msg.consumer.negative_acknowledge(msg.message)
def done_producing(self, consumer, orig_message, topic, result, sent_message):
if result == pulsar.Result.Ok:
@@ -269,6 +273,8 @@ class PythonInstance(object):
error_msg = "Failed to publish to topic [%s] with error [%s] with src message id [%s]" % (topic, result, orig_message.message_id())
Log.error(error_msg)
self.stats.incr_total_sys_exceptions(Exception(error_msg))
+ # If producer fails send output then send neg ack for input message back to broker
+ consumer.negative_acknowledge(orig_message)
def process_result(self, output, msg):
diff --git a/tests/docker-images/java-test-functions/pom.xml b/tests/docker-images/java-test-functions/pom.xml
new file mode 100644
index 0000000..68f1160
--- /dev/null
+++ b/tests/docker-images/java-test-functions/pom.xml
@@ -0,0 +1,81 @@
+<!--
+
+ 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.
+
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+ <parent>
+ <groupId>org.apache.pulsar.tests</groupId>
+ <artifactId>docker-images</artifactId>
+ <version>2.4.0-SNAPSHOT</version>
+ </parent>
+ <modelVersion>4.0.0</modelVersion>
+ <groupId>org.apache.pulsar.tests</groupId>
+ <artifactId>java-test-functions</artifactId>
+ <name>Apache Pulsar :: Tests :: Docker Images :: Java Test Functions</name>
+ <packaging>jar</packaging>
+
+ <profiles>
+ <profile>
+ <id>docker</id>
+ <activation>
+ <property>
+ <name>integrationTests</name>
+ </property>
+ </activation>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.pulsar</groupId>
+ <artifactId>pulsar-functions-api-examples</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+ </dependencies>
+
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-shade-plugin</artifactId>
+ <executions>
+ <execution>
+ <phase>package</phase>
+ <goals>
+ <goal>shade</goal>
+ </goals>
+ <configuration>
+ <!--Don't create dependency reduced pom. This causes maven to go into a infinite loop. May be a maven bug-->
+ <createDependencyReducedPom>false</createDependencyReducedPom>>
+ <transformers>
+ <transformer
+ implementation="org.apache.maven.plugins.shade.resource.ServicesResourceTransformer"/>
+ <transformer
+ implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
+ </transformer>
+ </transformers>
+ </configuration>
+ </execution>
+ </executions>
+ </plugin>
+ </plugins>
+ </build>
+ </profile>
+ </profiles>
+
+</project>
diff --git a/tests/docker-images/java-test-functions/src/main/java/org/apache/pulsar/tests/integration/functions/ExceptionFunction.java b/tests/docker-images/java-test-functions/src/main/java/org/apache/pulsar/tests/integration/functions/ExceptionFunction.java
new file mode 100644
index 0000000..c2a13e2
--- /dev/null
+++ b/tests/docker-images/java-test-functions/src/main/java/org/apache/pulsar/tests/integration/functions/ExceptionFunction.java
@@ -0,0 +1,35 @@
+/**
+ * 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.tests.integration.functions;
+
+import java.util.function.Function;
+
+public class ExceptionFunction implements Function<String, String> {
+
+ int i = 0;
+ @Override
+ public String apply(String s) {
+ i++;
+ if (i % 10 == 0) {
+ throw new RuntimeException("test");
+ }
+
+ return s + "!";
+ }
+}
diff --git a/tests/docker-images/latest-version-image/Dockerfile b/tests/docker-images/latest-version-image/Dockerfile
index ab7b7f8..be212c0 100644
--- a/tests/docker-images/latest-version-image/Dockerfile
+++ b/tests/docker-images/latest-version-image/Dockerfile
@@ -48,3 +48,5 @@ COPY python-examples/exclamation_with_extra_deps.py /pulsar/examples/python-exam
COPY python-examples/exclamation.zip /pulsar/examples/python-examples/
COPY python-examples/producer_schema.py /pulsar/examples/python-examples/
COPY python-examples/consumer_schema.py /pulsar/examples/python-examples/
+COPY python-examples/exception_function.py /pulsar/examples/python-examples/
+COPY target/java-test-functions.jar /pulsar/examples/
diff --git a/tests/docker-images/latest-version-image/pom.xml b/tests/docker-images/latest-version-image/pom.xml
index 2f3db04..5076110 100644
--- a/tests/docker-images/latest-version-image/pom.xml
+++ b/tests/docker-images/latest-version-image/pom.xml
@@ -41,6 +41,11 @@
</activation>
<dependencies>
<dependency>
+ <groupId>org.apache.pulsar.tests</groupId>
+ <artifactId>java-test-functions</artifactId>
+ <version>${project.parent.version}</version>
+ </dependency>
+ <dependency>
<groupId>org.apache.pulsar</groupId>
<artifactId>pulsar-all-docker-image</artifactId>
<version>${project.parent.version}</version>
@@ -50,6 +55,31 @@
<build>
<plugins>
<plugin>
+ <artifactId>maven-dependency-plugin</artifactId>
+ <executions>
+ <execution>
+ <id>copy-installed</id>
+ <phase>package</phase>
+ <goals>
+ <goal>copy</goal>
+ </goals>
+ <configuration>
+ <artifactItems>
+ <artifactItem>
+ <groupId>org.apache.pulsar.tests</groupId>
+ <artifactId>java-test-functions</artifactId>
+ <version>${project.parent.version}</version>
+ <type>jar</type>
+ <overWrite>true</overWrite>
+ <outputDirectory>${project.build.directory}</outputDirectory>
+ <destFileName>java-test-functions.jar</destFileName>
+ </artifactItem>
+ </artifactItems>
+ </configuration>
+ </execution>
+ </executions>
+ </plugin>
+ <plugin>
<groupId>com.spotify</groupId>
<artifactId>dockerfile-maven-plugin</artifactId>
<version>${dockerfile-maven.version}</version>
diff --git a/tests/docker-images/latest-version-image/python-examples/exception_function.py b/tests/docker-images/latest-version-image/python-examples/exception_function.py
new file mode 100644
index 0000000..b14f389
--- /dev/null
+++ b/tests/docker-images/latest-version-image/python-examples/exception_function.py
@@ -0,0 +1,29 @@
+#!/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.
+#
+
+
+i = 0
+def process(input):
+ global i
+ i += 1
+ if i % 10 == 0:
+ raise Exception("test");
+
+ return input + "!"
\ No newline at end of file
diff --git a/tests/docker-images/pom.xml b/tests/docker-images/pom.xml
index 3084e07..f65001b 100644
--- a/tests/docker-images/pom.xml
+++ b/tests/docker-images/pom.xml
@@ -32,6 +32,7 @@
<artifactId>docker-images</artifactId>
<name>Apache Pulsar :: Tests :: Docker Images</name>
<modules>
+ <module>java-test-functions</module>
<module>latest-version-image</module>
</modules>
</project>
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 bb4c6b4..c31329f 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
@@ -726,6 +726,182 @@ public abstract class PulsarFunctionsTest extends PulsarFunctionsTestBase {
//
@Test
+ public void testPythonFunctionNegAck() throws Exception {
+ testFunctionNegAck(Runtime.PYTHON);
+ }
+
+ @Test
+ public void testJavaFunctionNegAck() throws Exception {
+ testFunctionNegAck(Runtime.JAVA);
+ }
+
+ private void testFunctionNegAck(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-neg-ack-" + runtime + "-input-" + randomName(8);
+ String outputTopicName = "test-neg-ack-" + runtime + "-output-" + randomName(8);
+
+ String functionName = "test-neg-ack-fn-" + randomName(8);
+ final int numMessages = 20;
+
+ // submit the exclamation function
+
+ if (runtime == Runtime.PYTHON) {
+ submitFunction(
+ runtime, inputTopicName, outputTopicName, functionName, EXCEPTION_FUNCTION_PYTHON_FILE, EXCEPTION_PYTHON_CLASS, schema);
+ } else {
+ submitFunction(
+ runtime, inputTopicName, outputTopicName, functionName, null, EXCEPTION_JAVA_CLASS, schema);
+ }
+
+ // get function info
+ getFunctionInfoSuccess(functionName);
+
+ // get function stats
+ getFunctionStatsEmpty(functionName);
+
+ // publish and consume result
+ if (Runtime.JAVA == runtime) {
+ // java supports schema
+ @Cleanup PulsarClient client = PulsarClient.builder()
+ .serviceUrl(pulsarCluster.getPlainTextServiceUrl())
+ .build();
+ @Cleanup Consumer<String> consumer = client.newConsumer(Schema.STRING)
+ .topic(outputTopicName)
+ .subscriptionType(SubscriptionType.Exclusive)
+ .subscriptionName("test-sub")
+ .subscribe();
+ @Cleanup Producer<String> producer = client.newProducer(Schema.STRING)
+ .topic(inputTopicName)
+ .create();
+
+ for (int i = 0; i < numMessages; i++) {
+ producer.send("message-" + i);
+ }
+
+ Set<String> expectedMessages = new HashSet<>();
+ for (int i = 0; i < numMessages; i++) {
+ expectedMessages.add("message-" + i + "!");
+ }
+
+ for (int i = 0; i < numMessages; i++) {
+ Message<String> msg = consumer.receive(60 * 2, TimeUnit.SECONDS);
+ log.info("Received: {}", msg.getValue());
+ assertTrue(expectedMessages.contains(msg.getValue()));
+ expectedMessages.remove(msg.getValue());
+ }
+ assertEquals(expectedMessages.size(), 0);
+
+ } else {
+ // python doesn't support schema
+
+ @Cleanup PulsarClient client = PulsarClient.builder()
+ .serviceUrl(pulsarCluster.getPlainTextServiceUrl())
+ .build();
+ @Cleanup Consumer<byte[]> consumer = client.newConsumer(Schema.BYTES)
+ .topic(outputTopicName)
+ .subscriptionType(SubscriptionType.Exclusive)
+ .subscriptionName("test-sub")
+ .subscribe();
+
+ @Cleanup Producer<byte[]> producer = client.newProducer(Schema.BYTES)
+ .topic(inputTopicName)
+ .create();
+
+ for (int i = 0; i < numMessages; i++) {
+ producer.newMessage().value(("message-" + i).getBytes(UTF_8)).send();
+ }
+
+ Set<String> expectedMessages = new HashSet<>();
+ for (int i = 0; i < numMessages; i++) {
+ expectedMessages.add("message-" + i + "!");
+ }
+
+ for (int i = 0; i < numMessages; i++) {
+ Message<byte[]> msg = consumer.receive(60 * 2, TimeUnit.SECONDS);
+ String msgValue = new String(msg.getValue(), UTF_8);
+ log.info("Received: {}", msgValue);
+ assertTrue(expectedMessages.contains(msgValue));
+ expectedMessages.remove(msgValue);
+ }
+ assertEquals(expectedMessages.size(), 0);
+ }
+
+ // get function status
+ ContainerExecResult result = pulsarCluster.getAnyWorker().execCmd(
+ PulsarCluster.ADMIN_SCRIPT,
+ "functions",
+ "status",
+ "--tenant", "public",
+ "--namespace", "default",
+ "--name", functionName
+ );
+
+ FunctionStatus functionStatus = FunctionStatus.decode(result.getStdout());
+
+ assertEquals(functionStatus.getNumInstances(), 1);
+ assertEquals(functionStatus.getNumRunning(), 1);
+ assertEquals(functionStatus.getInstances().size(), 1);
+ assertEquals(functionStatus.getInstances().get(0).getInstanceId(), 0);
+ assertTrue(functionStatus.getInstances().get(0).getStatus().getAverageLatency() > 0.0);
+ assertEquals(functionStatus.getInstances().get(0).getStatus().isRunning(), true);
+ assertTrue(functionStatus.getInstances().get(0).getStatus().getLastInvocationTime() > 0);
+ // going to receive two more tuples because of delivery
+ assertEquals(functionStatus.getInstances().get(0).getStatus().getNumReceived(), numMessages + 2);
+ // only going to successfully process 20
+ assertEquals(functionStatus.getInstances().get(0).getStatus().getNumSuccessfullyProcessed(), numMessages);
+ assertEquals(functionStatus.getInstances().get(0).getStatus().getNumRestarts(), 0);
+ assertEquals(functionStatus.getInstances().get(0).getStatus().getLatestUserExceptions().size(), 2);
+ assertEquals(functionStatus.getInstances().get(0).getStatus().getLatestSystemExceptions().size(), 0);
+
+ // get function stats
+ result = pulsarCluster.getAnyWorker().execCmd(
+ PulsarCluster.ADMIN_SCRIPT,
+ "functions",
+ "stats",
+ "--tenant", "public",
+ "--namespace", "default",
+ "--name", functionName
+ );
+
+ log.info("FUNCTION STATS: {}", result.getStdout());
+
+ FunctionStats functionStats = FunctionStats.decode(result.getStdout());
+ assertEquals(functionStats.getReceivedTotal(), numMessages + 2);
+ assertEquals(functionStats.getProcessedSuccessfullyTotal(), numMessages);
+ assertEquals(functionStats.getSystemExceptionsTotal(), 0);
+ assertEquals(functionStats.getUserExceptionsTotal(), 2);
+ assertTrue(functionStats.avgProcessLatency > 0);
+ assertTrue(functionStats.getLastInvocation() > 0);
+
+ assertEquals(functionStats.instances.size(), 1);
+ assertEquals(functionStats.instances.get(0).getInstanceId(), 0);
+ assertEquals(functionStats.instances.get(0).getMetrics().getReceivedTotal(), numMessages + 2);
+ assertEquals(functionStats.instances.get(0).getMetrics().getProcessedSuccessfullyTotal(), numMessages);
+ assertEquals(functionStats.instances.get(0).getMetrics().getSystemExceptionsTotal(), 0);
+ assertEquals(functionStats.instances.get(0).getMetrics().getUserExceptionsTotal(), 2);
+ assertTrue(functionStats.instances.get(0).getMetrics().avgProcessLatency > 0);
+
+ // delete function
+ deleteFunction(functionName);
+
+ // get function info
+ getFunctionInfoNotFound(functionName);
+
+ // make sure subscriptions are cleanup
+ checkSubscriptionsCleanup(inputTopicName);
+ }
+
+ @Test
public void testPythonPublishFunction() throws Exception {
testPublishFunction(Runtime.PYTHON);
}
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 e717389..823b7d1 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
@@ -76,6 +76,9 @@ public abstract class PulsarFunctionsTestBase extends PulsarTestSuite {
public static final String PUBLISH_JAVA_CLASS =
"org.apache.pulsar.functions.api.examples.PublishFunctionWithMessageConf";
+ public static final String EXCEPTION_JAVA_CLASS =
+ "org.apache.pulsar.tests.integration.functions.ExceptionFunction";
+
public static final String EXCLAMATION_PYTHON_CLASS =
"exclamation_function.ExclamationFunction";
@@ -86,13 +89,15 @@ public abstract class PulsarFunctionsTestBase extends PulsarTestSuite {
public static final String EXCLAMATION_PYTHONZIP_CLASS =
"exclamation";
+ public static final String EXCEPTION_PYTHON_CLASS = "exception_function";
+
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";
-
+ public static final String EXCEPTION_FUNCTION_PYTHON_FILE = "exception_function.py";
protected static String getExclamationClass(Runtime runtime,
boolean pyZip,
diff --git a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/functions/utils/CommandGenerator.java b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/functions/utils/CommandGenerator.java
index 64b70a6..2b16b29 100644
--- a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/functions/utils/CommandGenerator.java
+++ b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/functions/utils/CommandGenerator.java
@@ -54,7 +54,7 @@ public class CommandGenerator {
private Long slidingIntervalDurationMs;
private Map<String, String> userConfig = new HashMap<>();
- private static final String JAVAJAR = "/pulsar/examples/api-examples.jar";
+ private static final String JAVAJAR = "/pulsar/examples/java-test-functions.jar";
private static final String PYTHONBASE = "/pulsar/examples/python-examples/";
public static CommandGenerator createDefaultGenerator(String sourceTopic, String functionClassName) {