From c65dcfa4f137fe2a0bbff2967f6b703d7dc4c3c0 Mon Sep 17 00:00:00 2001 From: david-streamlio <35466513+david-streamlio@users.noreply.github.com> Date: Fri, 21 Aug 2026 11:10:52 -0700 Subject: [PATCH] [fix][fn] Honour negativeAckRedeliveryDelayMs in the Go function runtime The Go runtime negatively acknowledges on failure but never configured the redelivery delay, so the client default of 60 seconds applied regardless of what SourceSpec.NegativeAckRedeliveryDelayMs carried. Add resolveNackRedeliveryDelay and set NackRedeliveryDelay on all four ConsumerOptions in setupConsumer - the regex and non-regex branches of both the receiver-queue-size and default paths. The field is a proto3 scalar with no presence, so an unset value reads as 0. Only a positive value produces a duration; zero leaves ConsumerOptions at its zero value, which the client treats as unset. That matches the guard the Java runtime applies in JavaInstanceRunnable, and means a function that does not configure the delay is unaffected. Fixes #26409 Master Issue: #26404 Co-Authored-By: Claude Opus 5 (1M context) --- pulsar-function-go/pf/instance.go | 65 +++++++++++++++--------- pulsar-function-go/pf/nackDelay_test.go | 66 +++++++++++++++++++++++++ 2 files changed, 109 insertions(+), 22 deletions(-) create mode 100644 pulsar-function-go/pf/nackDelay_test.go 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)) + }) + } +}