[fix][fn] Honour negativeAckRedeliveryDelayMs in the Python function runtime

The Python 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. A
function configured for fast retry, or for a long back-off from a
struggling downstream, silently got neither.

Add get_negative_ack_args() and splat its result into all three
subscribe() call sites.

Two details drive the shape:

- The field is a proto3 scalar with no presence, so an unset value reads
  as 0. Only a positive value is forwarded, leaving the client default in
  place otherwise - the same guard JavaInstanceRunnable applies. Sending 0
  through would mean immediate redelivery rather than the default.

- The argument is omitted rather than passed as None. subscribe()
  validates it with _check_type(int, ...) and not _check_type_or_none, so
  None would raise for every function that does not configure it, unlike
  the neighbouring unacked_messages_timeout_ms which does accept None.

Returning a dict to splat rather than a value keeps that omission at one
site, since the first call site passes explicit keywords while the other
two build a consumer_args dict.

Fixes #26411

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

Author: david-streamlio

pulsar-functions/instance/src/main/python/python_instance.py

@@ -150,6 +150,8 @@ def run(self):
     elif self.instance_config.function_details.retainKeyOrdering:
       mode = pulsar._pulsar.ConsumerType.KeyShared
 
+    nack_args = self.get_negative_ack_args()
+
     position = pulsar._pulsar.InitialPosition.Latest
     if self.instance_config.function_details.source.subscriptionPosition == Function_pb2.SubscriptionPosition.Value("EARLIEST"):
       position = pulsar._pulsar.InitialPosition.Earliest
@@ -181,7 +183,8 @@ def run(self):
         message_listener=partial(self.message_listener, self.input_serdes[topic], DEFAULT_SCHEMA),
         unacked_messages_timeout_ms=int(self.timeout_ms) if self.timeout_ms else None,
         initial_position=position,
-        properties=properties
+        properties=properties,
+        **nack_args
       )
 
     for topic, consumer_conf in self.instance_config.function_details.source.inputSpecs.items():
@@ -207,6 +210,7 @@ def run(self):
         "properties": properties,
         "crypto_key_reader": crypto_key_reader
       }
+      consumer_args.update(nack_args)
       if consumer_conf.HasField("receiverQueueSize"):
         consumer_args["receiver_queue_size"] = consumer_conf.receiverQueueSize.value
 
@@ -584,6 +588,26 @@ def get_record_class(self, class_name):
         except:
           pass
       return record_kclass
+  def get_negative_ack_args(self):
+    """Build the negative-ack redelivery delay argument for Client.subscribe().
+
+    Returns a dict to splat into the subscribe() call: either empty, or carrying
+    negative_ack_redelivery_delay_ms.
+
+    SourceSpec.negativeAckRedeliveryDelayMs is a proto3 scalar with no presence, so an unset field
+    reads as 0. Only a positive value is forwarded, leaving the client default (60s) in place
+    otherwise - the same guard the Java runtime applies in JavaInstanceRunnable.
+
+    The argument is omitted rather than passed as None because subscribe() validates it with
+    _check_type(int, ...) rather than _check_type_or_none, so None would fail for every function
+    that does not configure it.
+    """
+    delay_ms = self.instance_config.function_details.source.negativeAckRedeliveryDelayMs
+    if delay_ms <= 0:
+      return {}
+
+    return {"negative_ack_redelivery_delay_ms": delay_ms}
+
   def get_crypto_reader(self, crypto_spec):
     crypto_key_reader = None
     if crypto_spec is not None:

pulsar-functions/instance/src/test/python/test_python_instance.py

@@ -149,3 +149,39 @@ def test_do_not_forward_properties(self):
     self.assertNotIn("custom-key", kwargs['properties'])
     self.assertIn("__pfn_input_topic__", kwargs['properties'])
 
+
+class TestNegativeAckRedeliveryDelay(unittest.TestCase):
+  """Covers SourceSpec.negativeAckRedeliveryDelayMs reaching the consumer.
+
+  The runtime negatively acknowledges on failure but never configured the delay, so the client
+  default of 60s always applied. The Java runtime guards on > 0 in JavaInstanceRunnable.
+  """
+
+  def _instance(self, delay_ms=None):
+    function_details = Function_pb2.FunctionDetails()
+    function_details.sink.topic = "test_sink_topic"
+    if delay_ms is not None:
+      function_details.source.negativeAckRedeliveryDelayMs = delay_ms
+
+    return PythonInstance('test_instance', 'test_func', '1.0', function_details, 100, 30,
+                          'user_code', Mock(), Mock(), 'test_cluster', 'test_url', None)
+
+  def test_positive_delay_is_forwarded(self):
+    args = self._instance(delay_ms=5000).get_negative_ack_args()
+    self.assertEqual({"negative_ack_redelivery_delay_ms": 5000}, args)
+
+  def test_unset_delay_is_omitted(self):
+    # proto3 scalar with no presence: unset reads as 0. The argument must be omitted rather than
+    # sent - subscribe() validates it with _check_type(int), so None would fail for every function
+    # that does not set it, and 0 would mean immediate redelivery instead of the 60s default.
+    self.assertEqual({}, self._instance().get_negative_ack_args())
+
+  def test_explicit_zero_is_omitted(self):
+    self.assertEqual({}, self._instance(delay_ms=0).get_negative_ack_args())
+
+  def test_result_is_splattable_into_subscribe_kwargs(self):
+    # The value is consumed via **nack_args and consumer_args.update(...), so it must be a dict
+    # with exactly the keyword subscribe() expects.
+    args = self._instance(delay_ms=250).get_negative_ack_args()
+    self.assertIsInstance(args, dict)
+    self.assertEqual(["negative_ack_redelivery_delay_ms"], list(args.keys()))