diff --git a/pulsar-function-go/pf/instance.go b/pulsar-function-go/pf/instance.go index 49e156a4177fd..49164b98edff6 100644 --- a/pulsar-function-go/pf/instance.go +++ b/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 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)) + }) + } +}