From 6c95e4c0444cb35c4ba9df69f2456b4fd2c5d52e Mon Sep 17 00:00:00 2001 From: david-streamlio <35466513+david-streamlio@users.noreply.github.com> Date: Fri, 21 Aug 2026 11:09:09 -0700 Subject: [PATCH] [fix][fn] Honour retainOrdering and retainKeyOrdering in the Go function runtime setupConsumer picked the subscription type from SubscriptionType alone. RetainOrdering and RetainKeyOrdering appeared nowhere in pulsar-function-go outside the generated protobuf, so both were accepted by pulsar-admin, reported back by functions get, carried into the instance and then dropped. The consequence for retainKeyOrdering is a correctness one rather than a missing feature: the function ran on a Shared subscription, so messages sharing a key were distributed across instances and processed concurrently - exactly the guarantee the flag exists to provide. Nothing failed and nothing logged; the ordering was simply absent. Extract resolveSubscriptionType and apply both flags after the explicit SubscriptionType, matching the Java and Python runtimes. retainOrdering selects Failover, retainKeyOrdering selects KeyShared, and ordering wins when both are set - the precedence python_instance.py applies in its if/elif. EFFECTIVELY_ONCE needs no arm: instanceConf.go refuses it before an instance is built. The helper takes the three fields rather than FunctionDetails so it can be tested directly and so it does not copy a protobuf message by value, which go vet reports as copying a lock. Fixes #26405 Master Issue: #26404 --- pulsar-function-go/pf/instance.go | 28 +++++- .../pf/subscriptionType_test.go | 88 +++++++++++++++++++ 2 files changed, 114 insertions(+), 2 deletions(-) create mode 100644 pulsar-function-go/pf/subscriptionType_test.go diff --git a/pulsar-function-go/pf/instance.go b/pulsar-function-go/pf/instance.go index 2cdfc8a6e9497..6a174bf369d47 100644 --- a/pulsar-function-go/pf/instance.go +++ b/pulsar-function-go/pf/instance.go @@ -303,12 +303,36 @@ func (gi *goInstance) getProducer(topicName string) (pulsar.Producer, error) { return producer, err } -func (gi *goInstance) setupConsumer() (chan pulsar.ConsumerMessage, error) { +// 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(gi.context.instanceConf.funcDetails.Source.SubscriptionType) == pb.SubscriptionType_value["FAILOVER"] { + if int32(configured) == pb.SubscriptionType_value["FAILOVER"] { + subscriptionType = pulsar.Failover + } + + if retainOrdering { subscriptionType = pulsar.Failover + } else if retainKeyOrdering { + subscriptionType = pulsar.KeyShared } + return subscriptionType +} + +func (gi *goInstance) setupConsumer() (chan pulsar.ConsumerMessage, error) { + 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 if funcDetails.Source != nil && funcDetails.Source.SubscriptionName != "" { diff --git a/pulsar-function-go/pf/subscriptionType_test.go b/pulsar-function-go/pf/subscriptionType_test.go new file mode 100644 index 0000000000000..f2fc8bead04d1 --- /dev/null +++ b/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)) + }) + } +}