Function use negative acknowledge

pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/stats/FunctionStatsManager.java

@@ -437,8 +437,5 @@ public void reset() {
 
         statTotalRecordsReceived1min.clear();
         _statTotalRecordsReceived1min = statTotalRecordsReceived1min.labels(metricsLabels);
-
-        latestUserExceptions.clear();
-        latestSystemExceptions.clear();
     }
 }

pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/stats/SinkStatsManager.java

@@ -191,9 +191,6 @@ public void reset() {
 
         statTotalWritten1min.clear();
         _statTotalWritten1min = statTotalWritten1min.labels(metricsLabels);
-
-        latestSystemExceptions.clear();
-        latestSinkExceptions.clear();
     }
 
     @Override

pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/stats/SourceStatsManager.java

@@ -190,9 +190,6 @@ public void reset() {
 
         statTotalWritten1min.clear();
         _statTotalWritten1min = statTotalWritten1min.labels(metricsLabels);
-
-        latestSystemExceptions.clear();
-        latestSourceExceptions.clear();
     }
 
     @Override

pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/source/PulsarSource.java

@@ -122,6 +122,7 @@ public void received(Consumer<T> consumer, Message<T> message) {
                     if (pulsarSourceConfig.getProcessingGuarantees() == FunctionConfig.ProcessingGuarantees.EFFECTIVELY_ONCE) {
                         throw new RuntimeException("Failed to process message: " + message.getMessageId());
                     }
+                    consumer.negativeAcknowledge(message);
                 })
                 .build();
 

pulsar-functions/instance/src/main/python/function_stats.py

@@ -179,9 +179,9 @@ def set_last_invocation(self, time):
   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 @@ def report_system_exception_prometheus(self, exception, ts):
     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)

pulsar-functions/instance/src/main/python/python_instance.py

@@ -249,6 +249,8 @@ def actual_execution(self):
         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 @@ def actual_execution(self):
       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 @@ def done_producing(self, consumer, orig_message, topic, result, sent_message):
       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):

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 java-examples/target/java-test-functions.jar /pulsar/examples/

tests/docker-images/latest-version-image/java-examples/pom.xml

@@ -0,0 +1,69 @@
+<!--
+
+    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>latest-version-image</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 :: Latest Version Testing :: Java Test Functions</name>
+    <packaging>jar</packaging>
+
+    <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>
+                            <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>
+
+
+</project>

tests/docker-images/latest-version-image/java-examples/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 + "!";
+    }
+}

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 + "!"