[fix][fn] Go functions must retrieve consumers by non-particioned topic ID

pom.xml

@@ -181,7 +181,7 @@ flexible messaging model and an intuitive client API.</description>
     <joda.version>2.10.10</joda.version>
     <jclouds.version>2.5.0</jclouds.version>
     <guice.version>5.1.0</guice.version>
-    <sqlite-jdbc.version>3.36.0.3</sqlite-jdbc.version>
+    <sqlite-jdbc.version>3.42.0.0</sqlite-jdbc.version>
     <mysql-jdbc.version>8.0.11</mysql-jdbc.version>
     <postgresql-jdbc.version>42.5.1</postgresql-jdbc.version>
     <clickhouse-jdbc.version>0.4.6</clickhouse-jdbc.version>
@@ -284,7 +284,7 @@ flexible messaging model and an intuitive client API.</description>
     <maven-shade-plugin>3.4.1</maven-shade-plugin>
     <maven-antrun-plugin.version>3.1.0</maven-antrun-plugin.version>
     <properties-maven-plugin.version>1.1.0</properties-maven-plugin.version>
-    <nifi-nar-maven-plugin.version>1.3.4</nifi-nar-maven-plugin.version>
+    <nifi-nar-maven-plugin.version>1.5.0</nifi-nar-maven-plugin.version>
     <maven-checkstyle-plugin.version>3.1.2</maven-checkstyle-plugin.version>
     <git-commit-id-plugin.version>4.0.2</git-commit-id-plugin.version>
     <wagon-ssh-external.version>3.5.3</wagon-ssh-external.version>

pulsar-function-go/pf/instance.go

@@ -404,11 +404,25 @@ func (gi *goInstance) processResult(msgInput pulsar.Message, output []byte) {
 // ackInputMessage doesn't produce any result, or the user doesn't want the result.
 func (gi *goInstance) ackInputMessage(inputMessage pulsar.Message) {
 	log.Debugf("ack input message topic name is: %s", inputMessage.Topic())
-	gi.consumers[inputMessage.Topic()].Ack(inputMessage)
+	gi.respondMessage(inputMessage, true)
 }
 
 func (gi *goInstance) nackInputMessage(inputMessage pulsar.Message) {
-	gi.consumers[inputMessage.Topic()].Nack(inputMessage)
+	gi.respondMessage(inputMessage, false)
+}
+
+func (gi *goInstance) respondMessage(inputMessage pulsar.Message, ack bool) {
+	topicName, err := ParseTopicName(inputMessage.Topic())
+	if err != nil {
+		log.Errorf("unable respond to message ID %s - invalid topic: %v", messageIDStr(inputMessage), err)
+		return
+	}
+	// consumers are indexed by topic name only (no partition)
+	if ack {
+		gi.consumers[topicName.NameWithoutPartition()].Ack(inputMessage)
+		return
+	}
+	gi.consumers[topicName.NameWithoutPartition()].Nack(inputMessage)
 }
 
 func getIdleTimeout(timeoutMilliSecond time.Duration) time.Duration {

pulsar-function-go/pf/util.go

@@ -21,6 +21,8 @@ package pf
 
 import (
 	"fmt"
+
+	"github.com/apache/pulsar-client-go/pulsar"
 )
 
 func getProperties(fullyQualifiedName string, instanceID int) map[string]string {
@@ -39,3 +41,12 @@ func getDefaultSubscriptionName(tenant, namespace, name string) string {
 func getFullyQualifiedInstanceID(tenant, namespace, name string, instanceID int) string {
 	return fmt.Sprintf("%s/%s/%s:%d", tenant, namespace, name, instanceID)
 }
+
+func messageIDStr(msg pulsar.Message) string {
+	// <ledger ID>:<entry ID>:<partition index>:<batch index>
+	return fmt.Sprintf("%d:%d:%d:%d",
+		msg.ID().LedgerID(),
+		msg.ID().EntryID(),
+		msg.ID().PartitionIdx(),
+		msg.ID().BatchIdx())
+}