From 2c23d1a395bcc4c8b4a68d43f5bd0421d4ec0b3d Mon Sep 17 00:00:00 2001 From: david-streamlio <35466513+david-streamlio@users.noreply.github.com> Date: Fri, 21 Aug 2026 11:06:24 -0700 Subject: [PATCH] [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 --- .../src/main/python/python_instance.py | 26 +++++++++++++- .../src/test/python/test_python_instance.py | 36 +++++++++++++++++++ 2 files changed, 61 insertions(+), 1 deletion(-) diff --git a/pulsar-functions/instance/src/main/python/python_instance.py b/pulsar-functions/instance/src/main/python/python_instance.py index 5c57dfef79008..e02c83b289fc0 100755 --- a/pulsar-functions/instance/src/main/python/python_instance.py +++ b/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: diff --git a/pulsar-functions/instance/src/test/python/test_python_instance.py b/pulsar-functions/instance/src/test/python/test_python_instance.py index 1e72db8545816..ec49c6bf2fc26 100644 --- a/pulsar-functions/instance/src/test/python/test_python_instance.py +++ b/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()))