[fix][fn] Honour retainOrdering and retainKeyOrdering in the Go function runtime (#26414)

Author: david-streamlio

pulsar-function-go/pf/instance.go

@@ -303,6 +303,30 @@ func (gi *goInstance) getProducer(topicName string) (pulsar.Producer, error) {
 	return producer, err
 }
 
+// resolveSubscriptionType picks the consumer subscription type for the function.
+//
+// The ordering flags are applied after the explicit SubscriptionType, matching the Java and Python
+// runtimes: retainOrdering requires a single consumer per partition, so it selects Failover, and
+// retainKeyOrdering selects KeyShared. Ordering wins when both are set, which is the precedence
+// python_instance.py applies.
+//
+// EFFECTIVELY_ONCE needs no arm here: instanceConf.go refuses it before an instance is built.
+func resolveSubscriptionType(configured pb.SubscriptionType, retainOrdering,
+	retainKeyOrdering bool) pulsar.SubscriptionType {
+	subscriptionType := pulsar.Shared
+	if int32(configured) == pb.SubscriptionType_value["FAILOVER"] {
+		subscriptionType = pulsar.Failover
+	}
+
+	if retainOrdering {
+		subscriptionType = pulsar.Failover
+	} else if retainKeyOrdering {
+		subscriptionType = pulsar.KeyShared
+	}
+
+	return subscriptionType
+}
+
 // resolveNackRedeliveryDelay returns the negative-ack redelivery delay to apply, or zero to leave
 // the client default in place.
 //
@@ -319,10 +343,10 @@ func resolveNackRedeliveryDelay(delayMs uint64) time.Duration {
 }
 
 func (gi *goInstance) setupConsumer() (chan pulsar.ConsumerMessage, error) {
-	subscriptionType := pulsar.Shared
-	if int32(gi.context.instanceConf.funcDetails.Source.SubscriptionType) == pb.SubscriptionType_value["FAILOVER"] {
-		subscriptionType = pulsar.Failover
-	}
+	subscriptionType := resolveSubscriptionType(
+		gi.context.instanceConf.funcDetails.Source.SubscriptionType,
+		gi.context.instanceConf.funcDetails.RetainOrdering,
+		gi.context.instanceConf.funcDetails.RetainKeyOrdering)
 
 	funcDetails := gi.context.instanceConf.funcDetails
 	subscriptionName := funcDetails.Tenant + "/" + funcDetails.Namespace + "/" + funcDetails.Name

pulsar-function-go/pf/subscriptionType_test.go

@@ -0,0 +1,88 @@
+//
+// 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 pf
+
+import (
+	"testing"
+
+	"github.com/apache/pulsar-client-go/pulsar"
+	"github.com/stretchr/testify/assert"
+
+	pb "github.com/apache/pulsar/pulsar-function-go/pb"
+)
+
+// The Go runtime previously read only SubscriptionType, so retainOrdering and retainKeyOrdering
+// were accepted and silently dropped: a function created with --retain-key-ordering ran on a
+// Shared subscription and lost per-key ordering. These pin the rules the Java and Python runtimes
+// apply.
+func TestResolveSubscriptionType(t *testing.T) {
+	tests := []struct {
+		name              string
+		configured        pb.SubscriptionType
+		retainOrdering    bool
+		retainKeyOrdering bool
+		expected          pulsar.SubscriptionType
+	}{
+		{
+			name:       "default is shared",
+			configured: pb.SubscriptionType_SHARED,
+			expected:   pulsar.Shared,
+		},
+		{
+			name:       "explicit failover is honoured",
+			configured: pb.SubscriptionType_FAILOVER,
+			expected:   pulsar.Failover,
+		},
+		{
+			name:           "retainOrdering selects failover",
+			configured:     pb.SubscriptionType_SHARED,
+			retainOrdering: true,
+			expected:       pulsar.Failover,
+		},
+		{
+			name:              "retainKeyOrdering selects key_shared",
+			configured:        pb.SubscriptionType_SHARED,
+			retainKeyOrdering: true,
+			expected:          pulsar.KeyShared,
+		},
+		{
+			// python_instance.py applies retainOrdering first and only falls to retainKeyOrdering
+			// in the else branch, so ordering wins. Pinned so the two runtimes cannot drift.
+			name:              "retainOrdering wins over retainKeyOrdering",
+			configured:        pb.SubscriptionType_SHARED,
+			retainOrdering:    true,
+			retainKeyOrdering: true,
+			expected:          pulsar.Failover,
+		},
+		{
+			name:              "retainKeyOrdering overrides an explicit failover",
+			configured:        pb.SubscriptionType_FAILOVER,
+			retainKeyOrdering: true,
+			expected:          pulsar.KeyShared,
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			assert.Equal(t, test.expected,
+				resolveSubscriptionType(test.configured, test.retainOrdering, test.retainKeyOrdering))
+		})
+	}
+}