diff --git a/pulsar-function-go/pf/instance.go b/pulsar-function-go/pf/instance.go index 2cdfc8a6e9497..49e156a4177fd 100644 --- a/pulsar-function-go/pf/instance.go +++ b/pulsar-function-go/pf/instance.go @@ -303,6 +303,21 @@ func (gi *goInstance) getProducer(topicName string) (pulsar.Producer, error) { return producer, err } +// resolveNackRedeliveryDelay returns the negative-ack redelivery delay to apply, or zero to leave +// the client default in place. +// +// SourceSpec.NegativeAckRedeliveryDelayMs is a proto3 scalar with no presence, so an unset field +// reads as 0. Only a positive value is applied, matching the guard the Java runtime uses in +// JavaInstanceRunnable; a zero left in ConsumerOptions is treated by the client as unset, so the +// default applies either way. +func resolveNackRedeliveryDelay(delayMs uint64) time.Duration { + if delayMs == 0 { + return 0 + } + + return time.Duration(delayMs) * time.Millisecond +} + func (gi *goInstance) setupConsumer() (chan pulsar.ConsumerMessage, error) { subscriptionType := pulsar.Shared if int32(gi.context.instanceConf.funcDetails.Source.SubscriptionType) == pb.SubscriptionType_value["FAILOVER"] { @@ -320,6 +335,8 @@ func (gi *goInstance) setupConsumer() (chan pulsar.ConsumerMessage, error) { funcDetails.Namespace, funcDetails.Name), gi.context.instanceConf.instanceID) + nackRedeliveryDelay := resolveNackRedeliveryDelay(funcDetails.Source.NegativeAckRedeliveryDelayMs) + channel := make(chan pulsar.ConsumerMessage) var ( @@ -338,39 +355,43 @@ func (gi *goInstance) setupConsumer() (chan pulsar.ConsumerMessage, error) { if consumerConf.ReceiverQueueSize != nil { if consumerConf.IsRegexPattern { consumer, err = gi.client.Subscribe(pulsar.ConsumerOptions{ - TopicsPattern: topicName.Name, - ReceiverQueueSize: int(consumerConf.ReceiverQueueSize.Value), - SubscriptionName: subscriptionName, - Properties: properties, - Type: subscriptionType, - MessageChannel: channel, + TopicsPattern: topicName.Name, + ReceiverQueueSize: int(consumerConf.ReceiverQueueSize.Value), + SubscriptionName: subscriptionName, + Properties: properties, + Type: subscriptionType, + MessageChannel: channel, + NackRedeliveryDelay: nackRedeliveryDelay, }) } else { consumer, err = gi.client.Subscribe(pulsar.ConsumerOptions{ - Topic: topicName.Name, - SubscriptionName: subscriptionName, - Properties: properties, - Type: subscriptionType, - ReceiverQueueSize: int(consumerConf.ReceiverQueueSize.Value), - MessageChannel: channel, + Topic: topicName.Name, + SubscriptionName: subscriptionName, + Properties: properties, + Type: subscriptionType, + ReceiverQueueSize: int(consumerConf.ReceiverQueueSize.Value), + MessageChannel: channel, + NackRedeliveryDelay: nackRedeliveryDelay, }) } } else { if consumerConf.IsRegexPattern { consumer, err = gi.client.Subscribe(pulsar.ConsumerOptions{ - TopicsPattern: topicName.Name, - SubscriptionName: subscriptionName, - Properties: properties, - Type: subscriptionType, - MessageChannel: channel, + TopicsPattern: topicName.Name, + SubscriptionName: subscriptionName, + Properties: properties, + Type: subscriptionType, + MessageChannel: channel, + NackRedeliveryDelay: nackRedeliveryDelay, }) } else { consumer, err = gi.client.Subscribe(pulsar.ConsumerOptions{ - Topic: topicName.Name, - SubscriptionName: subscriptionName, - Properties: properties, - Type: subscriptionType, - MessageChannel: channel, + Topic: topicName.Name, + SubscriptionName: subscriptionName, + Properties: properties, + Type: subscriptionType, + MessageChannel: channel, + NackRedeliveryDelay: nackRedeliveryDelay, }) } diff --git a/pulsar-function-go/pf/nackDelay_test.go b/pulsar-function-go/pf/nackDelay_test.go new file mode 100644 index 0000000000000..0ec704eca7e4c --- /dev/null +++ b/pulsar-function-go/pf/nackDelay_test.go @@ -0,0 +1,66 @@ +// +// 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" + "time" + + "github.com/stretchr/testify/assert" +) + +// The Go runtime nacks on failure but never configured the delay, so the client default of 60s +// applied regardless of SourceSpec.NegativeAckRedeliveryDelayMs. +func TestResolveNackRedeliveryDelay(t *testing.T) { + tests := []struct { + name string + delayMs uint64 + expected time.Duration + }{ + { + // proto3 scalar with no presence: unset reads as 0. Returning zero leaves + // ConsumerOptions at its zero value, which the client treats as unset. + name: "unset leaves the client default", + delayMs: 0, + expected: 0, + }, + { + name: "milliseconds are converted to a duration", + delayMs: 5000, + expected: 5 * time.Second, + }, + { + name: "sub-second values are preserved", + delayMs: 250, + expected: 250 * time.Millisecond, + }, + { + name: "the client default expressed explicitly still round-trips", + delayMs: 60000, + expected: time.Minute, + }, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + assert.Equal(t, test.expected, resolveNackRedeliveryDelay(test.delayMs)) + }) + } +}