[fix][fn] Honour negativeAckRedeliveryDelayMs in the Go function runtime (#26415)

(cherry picked from commit 12b86b1)

Author: david-streamlio

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,
 				})
 
 			}

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))
+		})
+	}
+}