From 5cc6207b06113afd97a7ec86b8202d0c6d155857 Mon Sep 17 00:00:00 2001 From: david-streamlio <35466513+david-streamlio@users.noreply.github.com> Date: Thu, 20 Aug 2026 17:33:55 -0700 Subject: [PATCH 1/6] [fix][fn] Honour deadLetterTopic and maxMessageRetries in the Python runtime FunctionConfig accepts maxMessageRetries and deadLetterTopic, both are carried into the instance as FunctionDetails.retryDetails, and the Java runtime applies them. The Python runtime ignored them entirely: nothing in python_instance.py referenced retryDetails, so a function created with --dead-letter-topic was accepted, reported back faithfully by functions get, and then routed nothing to the DLQ at runtime. Build a ConsumerDeadLetterPolicy from retryDetails in a new get_dead_letter_policy() and pass it to all three subscribe() call sites. The rules mirror the Java runtime, which guards on hasRetryDetails() in JavaInstanceRunnable and applies the policy in PulsarSource, setting the dead letter topic only when it is non-empty so the client can derive its "--DLQ" default. Two cases cannot mirror Java exactly, and both warn rather than failing the instance or silently doing nothing: - Java accepts maxMessageRetries >= 0, but the Python client's ConsumerDeadLetterPolicy rejects a redelivery count below 1, so zero cannot be expressed. Attaching no policy is the only option; a warning names the dead letter topic that will not receive messages. - A dead letter policy only takes effect on Shared and KeyShared subscriptions. retainOrdering and EFFECTIVELY_ONCE both select Failover, where the policy would be silently ineffective, so that combination warns too. Silently ineffective configuration is the bug this fixes, and it should not be reintroduced by the fix. Fixes #26397 --- .../src/main/python/python_instance.py | 46 +++++++++++- .../src/test/python/test_python_instance.py | 74 +++++++++++++++++++ 2 files changed, 118 insertions(+), 2 deletions(-) diff --git a/pulsar-functions/instance/src/main/python/python_instance.py b/pulsar-functions/instance/src/main/python/python_instance.py index 5c57dfef79008..946b11781aa8a 100755 --- a/pulsar-functions/instance/src/main/python/python_instance.py +++ b/pulsar-functions/instance/src/main/python/python_instance.py @@ -154,6 +154,8 @@ def run(self): if self.instance_config.function_details.source.subscriptionPosition == Function_pb2.SubscriptionPosition.Value("EARLIEST"): position = pulsar._pulsar.InitialPosition.Earliest + dead_letter_policy = self.get_dead_letter_policy(mode) + subscription_name = self.instance_config.function_details.source.subscriptionName if not (subscription_name and subscription_name.strip()): @@ -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, + dead_letter_policy=dead_letter_policy ) for topic, consumer_conf in self.instance_config.function_details.source.inputSpecs.items(): @@ -205,7 +208,8 @@ def run(self): "unacked_messages_timeout_ms": int(self.timeout_ms) if self.timeout_ms else None, "initial_position": position, "properties": properties, - "crypto_key_reader": crypto_key_reader + "crypto_key_reader": crypto_key_reader, + "dead_letter_policy": dead_letter_policy } if consumer_conf.HasField("receiverQueueSize"): consumer_args["receiver_queue_size"] = consumer_conf.receiverQueueSize.value @@ -584,6 +588,44 @@ def get_record_class(self, class_name): except: pass return record_kclass + def get_dead_letter_policy(self, consumer_type): + """Build the consumer dead letter policy from FunctionDetails.retryDetails. + + Mirrors the Java runtime (JavaInstanceRunnable + PulsarSource): the policy is only considered + when retryDetails is present, and an empty deadLetterTopic is left to the client, which defaults + it to "--DLQ". + + Returns None when no policy should be attached. + """ + if not self.instance_config.function_details.HasField("retryDetails"): + return None + + retry_details = self.instance_config.function_details.retryDetails + max_message_retries = retry_details.maxMessageRetries + + # The Java runtime accepts maxMessageRetries >= 0, but the Python client rejects a + # maxRedeliverCount below 1, so zero cannot be expressed here. Warn rather than fail the + # instance, and rather than dropping it silently - silent drops are the bug this fixes. + if max_message_retries < 1: + if max_message_retries == 0 and retry_details.deadLetterTopic: + Log.warning( + "maxMessageRetries is 0, which the Python client cannot express (it requires a " + "redelivery count of at least 1); no dead letter policy will be applied and messages " + "will not be routed to %s" % retry_details.deadLetterTopic) + return None + + # A dead letter policy only takes effect on Shared and KeyShared subscriptions. + if consumer_type not in (pulsar._pulsar.ConsumerType.Shared, pulsar._pulsar.ConsumerType.KeyShared): + Log.warning( + "a dead letter policy is configured but the subscription type is not Shared or " + "KeyShared, so it will have no effect; retainOrdering and EFFECTIVELY_ONCE both select " + "a Failover subscription") + return None + + return pulsar.ConsumerDeadLetterPolicy( + max_redeliver_count=max_message_retries, + dead_letter_topic=retry_details.deadLetterTopic or None) + 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..e945fd937a135 100644 --- a/pulsar-functions/instance/src/test/python/test_python_instance.py +++ b/pulsar-functions/instance/src/test/python/test_python_instance.py @@ -32,6 +32,7 @@ from contextimpl import ContextImpl from python_instance import PythonInstance, InstanceConfig +import pulsar from pulsar import Message import Function_pb2 @@ -149,3 +150,76 @@ def test_do_not_forward_properties(self): self.assertNotIn("custom-key", kwargs['properties']) self.assertIn("__pfn_input_topic__", kwargs['properties']) + +class TestDeadLetterPolicy(unittest.TestCase): + """Covers FunctionDetails.retryDetails -> ConsumerDeadLetterPolicy. + + The Java runtime applies these in JavaInstanceRunnable (guarded on hasRetryDetails) and + PulsarSource (maxMessageRetries >= 0, deadLetterTopic only when non-empty). The Python runtime + previously ignored retryDetails entirely. + """ + + def _instance(self, max_message_retries=None, dead_letter_topic=None): + function_details = Function_pb2.FunctionDetails() + function_details.sink.topic = "test_sink_topic" + if max_message_retries is not None: + function_details.retryDetails.maxMessageRetries = max_message_retries + if dead_letter_topic is not None: + function_details.retryDetails.deadLetterTopic = dead_letter_topic + + return PythonInstance('test_instance', 'test_func', '1.0', function_details, 100, 30, + 'user_code', Mock(), Mock(), 'test_cluster', 'test_url', None) + + def test_no_retry_details_means_no_policy(self): + instance = self._instance() + self.assertIsNone( + instance.get_dead_letter_policy(pulsar._pulsar.ConsumerType.Shared)) + + def test_policy_built_from_retry_details(self): + instance = self._instance(max_message_retries=3, + dead_letter_topic="persistent://public/default/my-dlq") + policy = instance.get_dead_letter_policy(pulsar._pulsar.ConsumerType.Shared) + + self.assertIsNotNone(policy) + self.assertEqual(3, policy.max_redeliver_count) + self.assertEqual("persistent://public/default/my-dlq", policy.dead_letter_topic) + + def test_empty_dead_letter_topic_defers_to_client_default(self): + # The Java runtime only sets the topic when non-empty, leaving the client to derive + # "--DLQ". Passing "" through would override that with an invalid name. + instance = self._instance(max_message_retries=2) + policy = instance.get_dead_letter_policy(pulsar._pulsar.ConsumerType.Shared) + + self.assertIsNotNone(policy) + self.assertEqual(2, policy.max_redeliver_count) + + def test_zero_retries_attaches_no_policy(self): + # Java accepts maxMessageRetries >= 0, but ConsumerDeadLetterPolicy rejects a redelivery count + # below 1, so zero cannot be expressed here. It must not raise and take the instance down. + instance = self._instance(max_message_retries=0, + dead_letter_topic="persistent://public/default/my-dlq") + self.assertIsNone( + instance.get_dead_letter_policy(pulsar._pulsar.ConsumerType.Shared)) + + def test_negative_retries_attaches_no_policy(self): + instance = self._instance(max_message_retries=-1) + self.assertIsNone( + instance.get_dead_letter_policy(pulsar._pulsar.ConsumerType.Shared)) + + def test_key_shared_subscription_gets_policy(self): + instance = self._instance(max_message_retries=3) + self.assertIsNotNone( + instance.get_dead_letter_policy(pulsar._pulsar.ConsumerType.KeyShared)) + + def test_failover_subscription_gets_no_policy(self): + # A dead letter policy has no effect on Failover, which retainOrdering and EFFECTIVELY_ONCE + # both select. Returning None keeps that explicit rather than silently ineffective. + instance = self._instance(max_message_retries=3, + dead_letter_topic="persistent://public/default/my-dlq") + self.assertIsNone( + instance.get_dead_letter_policy(pulsar._pulsar.ConsumerType.Failover)) + + def test_exclusive_subscription_gets_no_policy(self): + instance = self._instance(max_message_retries=3) + self.assertIsNone( + instance.get_dead_letter_policy(pulsar._pulsar.ConsumerType.Exclusive)) From d2b5c838804cc2770029a3fe1b86353269135454 Mon Sep 17 00:00:00 2001 From: david-streamlio <35466513+david-streamlio@users.noreply.github.com> Date: Mon, 24 Aug 2026 16:12:49 -0700 Subject: [PATCH 2/6] [feat][fn] Allow message retries and a dead letter topic on Python functions ### Motivation `doPythonChecks` refuses any `maxMessageRetries >= 0`, so `pulsar-admin functions create --py ... --max-message-retries 3` fails at creation with "Message retries not yet supported in python". That guard is now the only thing standing between the Python runtime and a working dead letter queue. The runtime honours `FunctionDetails.retryDetails`, but nothing can reach it through the cluster path, because the two gates line up exactly: - `FunctionConfigUtils.convert` only populates `retryDetails` when `maxMessageRetries != null && >= 0` -- the condition `doPythonChecks` rejects. So `--dead-letter-topic` on its own never produces a `retryDetails` message at all, and the runtime sees nothing to honour. - `--max-message-retries` with any value the runtime could use is refused before it gets that far. `validateNonJavaFunction` has one caller, the worker REST API (`FunctionsImpl`), so the refusal applies to cluster submission only; `LocalRunner` never calls it, which is why the runtime path is reachable under `localrun` today and nowhere else. ### Modifications Replace the blanket refusal in `doPythonChecks` with a narrow one on zero. Zero asks for no redelivery at all before the dead letter topic, and the Python client cannot express it: `ConsumerDeadLetterPolicy` requires a `maxRedeliverCount` of at least 1. Accepting it would create a function whose dead letter topic never receives anything -- the silently ineffective configuration this support exists to remove -- so it is rejected at creation, where the mistake is still cheap to fix, rather than warned about in an instance log nobody reads. A negative value leaves retries unset, as on the Java path. `doGolangChecks` keeps its guard: the Go runtime does not honour `retryDetails` yet, and a test now pins that so this change cannot be widened to Go by accident. Four tests: Python accepts retries with and without an explicit dead letter topic, rejects zero with the new message, and Go still refuses retries. --- .../functions/utils/FunctionConfigUtils.java | 11 ++++- .../utils/FunctionConfigUtilsTest.java | 43 +++++++++++++++++++ 2 files changed, 52 insertions(+), 2 deletions(-) diff --git a/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/FunctionConfigUtils.java b/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/FunctionConfigUtils.java index d88ac2820a966..1ff7cd795c19c 100644 --- a/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/FunctionConfigUtils.java +++ b/pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/FunctionConfigUtils.java @@ -757,8 +757,15 @@ private static void doPythonChecks(FunctionConfig functionConfig) { throw new IllegalArgumentException("There is currently no support windowing in python"); } - if (functionConfig.getMaxMessageRetries() != null && functionConfig.getMaxMessageRetries() >= 0) { - throw new IllegalArgumentException("Message retries not yet supported in python"); + // The Python runtime honours FunctionDetails.retryDetails, so message retries and a dead letter + // topic are no longer refused outright. Zero is still refused: it asks for no redelivery at all + // before the dead letter topic, and the Python client cannot express that -- ConsumerDeadLetterPolicy + // requires a maxRedeliverCount of at least 1. Accepting it would create a function whose dead letter + // topic never receives anything, which is the silently ineffective configuration this support was + // added to remove. A negative value leaves retries unset, as it does on the Java path. + if (functionConfig.getMaxMessageRetries() != null && functionConfig.getMaxMessageRetries() == 0) { + throw new IllegalArgumentException("maxMessageRetries must be at least 1 in python; the Python " + + "client cannot express a redelivery count of 0"); } } diff --git a/pulsar-functions/utils/src/test/java/org/apache/pulsar/functions/utils/FunctionConfigUtilsTest.java b/pulsar-functions/utils/src/test/java/org/apache/pulsar/functions/utils/FunctionConfigUtilsTest.java index ba5e40429cf2d..d4fe475135a89 100644 --- a/pulsar-functions/utils/src/test/java/org/apache/pulsar/functions/utils/FunctionConfigUtilsTest.java +++ b/pulsar-functions/utils/src/test/java/org/apache/pulsar/functions/utils/FunctionConfigUtilsTest.java @@ -533,6 +533,49 @@ public void testMergeRuntimeFlags() { } @SuppressWarnings("deprecation") + @Test + public void testPythonFunctionAcceptsMessageRetriesAndDeadLetterTopic() { + FunctionConfig functionConfig = createPythonFunctionConfig(); + functionConfig.setMaxMessageRetries(3); + functionConfig.setDeadLetterTopic("test-dlq"); + FunctionConfigUtils.validateNonJavaFunction(functionConfig); + } + + @Test + public void testPythonFunctionAcceptsMessageRetriesWithoutADeadLetterTopic() { + // The client defaults the topic to "--DLQ" when it is left empty. + FunctionConfig functionConfig = createPythonFunctionConfig(); + functionConfig.setMaxMessageRetries(1); + FunctionConfigUtils.validateNonJavaFunction(functionConfig); + } + + @Test(expectedExceptions = IllegalArgumentException.class, + expectedExceptionsMessageRegExp = "maxMessageRetries must be at least 1 in python.*") + public void testPythonFunctionRejectsZeroMessageRetries() { + // Zero asks for no redelivery before the dead letter topic, which the Python client cannot + // express, so the dead letter topic would never receive anything. + FunctionConfig functionConfig = createPythonFunctionConfig(); + functionConfig.setMaxMessageRetries(0); + functionConfig.setDeadLetterTopic("test-dlq"); + FunctionConfigUtils.validateNonJavaFunction(functionConfig); + } + + @Test(expectedExceptions = IllegalArgumentException.class, + expectedExceptionsMessageRegExp = "Message retries not yet supported in Go function") + public void testGoFunctionStillRejectsMessageRetries() { + // The Go runtime does not honour retryDetails yet, so its guard stays until it does. + FunctionConfig functionConfig = createPythonFunctionConfig(); + functionConfig.setRuntime(FunctionConfig.Runtime.GO); + functionConfig.setMaxMessageRetries(3); + FunctionConfigUtils.validateNonJavaFunction(functionConfig); + } + + private FunctionConfig createPythonFunctionConfig() { + FunctionConfig functionConfig = createFunctionConfig(); + functionConfig.setRuntime(FunctionConfig.Runtime.PYTHON); + return functionConfig; + } + private FunctionConfig createFunctionConfig() { FunctionConfig functionConfig = new FunctionConfig(); functionConfig.setTenant("test-tenant"); From e6367f4a78fa9d8d943b77f1fe8dd0f75b4425ef Mon Sep 17 00:00:00 2001 From: david-streamlio <35466513+david-streamlio@users.noreply.github.com> Date: Tue, 25 Aug 2026 07:29:35 -0700 Subject: [PATCH 3/6] [fix][fn] Separate get_dead_letter_policy with a blank line Every other method in python_instance.py is preceded by a blank line; get_dead_letter_policy, added earlier in this branch, ran on directly from the end of get_record_class. --- pulsar-functions/instance/src/main/python/python_instance.py | 1 + 1 file changed, 1 insertion(+) diff --git a/pulsar-functions/instance/src/main/python/python_instance.py b/pulsar-functions/instance/src/main/python/python_instance.py index 946b11781aa8a..1abee32b4d2b2 100755 --- a/pulsar-functions/instance/src/main/python/python_instance.py +++ b/pulsar-functions/instance/src/main/python/python_instance.py @@ -588,6 +588,7 @@ def get_record_class(self, class_name): except: pass return record_kclass + def get_dead_letter_policy(self, consumer_type): """Build the consumer dead letter policy from FunctionDetails.retryDetails. From 697cf6cbc0608c7fdeafddf8c280a05381d56cd0 Mon Sep 17 00:00:00 2001 From: david-streamlio <35466513+david-streamlio@users.noreply.github.com> Date: Tue, 25 Aug 2026 07:45:19 -0700 Subject: [PATCH 4/6] [fix][fn] Mark --max-message-retries and --dead-letter-topic as Python-capable The @Option descriptions in CmdFunctions carry a runtime marker that the docs sync parses into the "Support" column of the published pulsar-admin CLI reference, and that `functions create --help` prints verbatim. Both flags were marked #Java, which this branch makes untrue. --- .../main/java/org/apache/pulsar/admin/cli/CmdFunctions.java | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdFunctions.java b/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdFunctions.java index ce3d8d323685f..1c993e56eab1c 100644 --- a/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdFunctions.java +++ b/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdFunctions.java @@ -370,7 +370,7 @@ abstract class FunctionDetailsCommand extends BaseCommand { @Option(names = "--timeout-ms", description = "The message timeout in milliseconds #Java, Python") protected Long timeoutMs; @Option(names = "--max-message-retries", - description = "How many times should we try to process a message before giving up #Java") + description = "How many times should we try to process a message before giving up #Java, Python") protected Integer maxMessageRetries; @Option(names = "--custom-runtime-options", description = "A string that encodes options to " + "customize the runtime, see docs for configured runtime for details #Java") @@ -379,7 +379,7 @@ abstract class FunctionDetailsCommand extends BaseCommand { + "how the secret is fetched by the underlying secrets provider #Java, Python") protected String secretsString; @Option(names = "--dead-letter-topic", - description = "The topic where messages that are not processed successfully are sent to #Java") + description = "The topic where messages that are not processed successfully are sent to #Java, Python") protected String deadLetterTopic; @Option(names = "--runtime-flags", description = "Any flags that you want to pass to a runtime" + " (for process & Kubernetes runtime only).") From 47df72c0e0ce0ef65dcf01c172a6a582cf7505d5 Mon Sep 17 00:00:00 2001 From: david-streamlio <35466513+david-streamlio@users.noreply.github.com> Date: Wed, 26 Aug 2026 10:11:25 -0700 Subject: [PATCH 5/6] [fix][fn] Align the Python dead letter policy with the Java runtime Motivation: Review of #26400 found that the two guards in get_dead_letter_policy made the Python runtime behave differently from Java and put a support matrix in the runtime that belongs to the client. Modifications: - Pass maxMessageRetries through instead of returning None below 1. Java does not start with that value either: PulsarSource builds a policy for any maxMessageRetries >= 0 and ConsumerBuilderImpl.deadLetterPolicy then rejects "MaxRedeliverCount must be > 0". Because LocalRunner bypasses validateNonJavaFunction, the previous guard let localrun start with retries silently disabled while Java failed fast. ConsumerDeadLetterPolicy raises ValueError for 0 and for negatives, so both now fail the instance. - Drop the Shared/KeyShared gate and the consumer_type parameter. The Java runtime always forwards a configured DeadLetterPolicy; which subscription types can act on one is a client concern, and encoding the current native client limitation here would drift from the client over time. Verification: - Reworked TestDeadLetterPolicy: the zero and negative cases now assert the fail-fast, the three subscription-type cases are replaced by one asserting the policy is not gated on subscription type - run_python_instance_tests.sh equivalent passes: 10 tests, all green The cluster path keeps its earlier, friendlier diagnostic - validateNonJavaFunction still rejects maxMessageRetries == 0 for Python before submission, so this only changes what happens when that check is bypassed. --- .../src/main/python/python_instance.py | 32 ++++-------- .../src/test/python/test_python_instance.py | 51 +++++++++---------- 2 files changed, 34 insertions(+), 49 deletions(-) diff --git a/pulsar-functions/instance/src/main/python/python_instance.py b/pulsar-functions/instance/src/main/python/python_instance.py index 1abee32b4d2b2..ae1dc912ce0ab 100755 --- a/pulsar-functions/instance/src/main/python/python_instance.py +++ b/pulsar-functions/instance/src/main/python/python_instance.py @@ -154,7 +154,7 @@ def run(self): if self.instance_config.function_details.source.subscriptionPosition == Function_pb2.SubscriptionPosition.Value("EARLIEST"): position = pulsar._pulsar.InitialPosition.Earliest - dead_letter_policy = self.get_dead_letter_policy(mode) + dead_letter_policy = self.get_dead_letter_policy() subscription_name = self.instance_config.function_details.source.subscriptionName @@ -589,42 +589,28 @@ def get_record_class(self, class_name): pass return record_kclass - def get_dead_letter_policy(self, consumer_type): + def get_dead_letter_policy(self): """Build the consumer dead letter policy from FunctionDetails.retryDetails. Mirrors the Java runtime (JavaInstanceRunnable + PulsarSource): the policy is only considered when retryDetails is present, and an empty deadLetterTopic is left to the client, which defaults it to "--DLQ". + The configured value is passed through unchanged. Java forwards it the same way - PulsarSource + builds the policy for any maxMessageRetries >= 0 and ConsumerBuilderImpl.deadLetterPolicy then + rejects a redelivery count below 1 - so an unusable value fails the instance here too rather + than starting with retries quietly disabled. Whether a subscription type can act on the policy + is left to the client, as it is for Java. + Returns None when no policy should be attached. """ if not self.instance_config.function_details.HasField("retryDetails"): return None retry_details = self.instance_config.function_details.retryDetails - max_message_retries = retry_details.maxMessageRetries - - # The Java runtime accepts maxMessageRetries >= 0, but the Python client rejects a - # maxRedeliverCount below 1, so zero cannot be expressed here. Warn rather than fail the - # instance, and rather than dropping it silently - silent drops are the bug this fixes. - if max_message_retries < 1: - if max_message_retries == 0 and retry_details.deadLetterTopic: - Log.warning( - "maxMessageRetries is 0, which the Python client cannot express (it requires a " - "redelivery count of at least 1); no dead letter policy will be applied and messages " - "will not be routed to %s" % retry_details.deadLetterTopic) - return None - - # A dead letter policy only takes effect on Shared and KeyShared subscriptions. - if consumer_type not in (pulsar._pulsar.ConsumerType.Shared, pulsar._pulsar.ConsumerType.KeyShared): - Log.warning( - "a dead letter policy is configured but the subscription type is not Shared or " - "KeyShared, so it will have no effect; retainOrdering and EFFECTIVELY_ONCE both select " - "a Failover subscription") - return None return pulsar.ConsumerDeadLetterPolicy( - max_redeliver_count=max_message_retries, + max_redeliver_count=retry_details.maxMessageRetries, dead_letter_topic=retry_details.deadLetterTopic or None) def get_crypto_reader(self, crypto_spec): 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 e945fd937a135..09d8f91715e69 100644 --- a/pulsar-functions/instance/src/test/python/test_python_instance.py +++ b/pulsar-functions/instance/src/test/python/test_python_instance.py @@ -157,6 +157,11 @@ class TestDeadLetterPolicy(unittest.TestCase): The Java runtime applies these in JavaInstanceRunnable (guarded on hasRetryDetails) and PulsarSource (maxMessageRetries >= 0, deadLetterTopic only when non-empty). The Python runtime previously ignored retryDetails entirely. + + The configured redelivery count is passed straight through, matching Java: PulsarSource builds a + policy for any value >= 0 and ConsumerBuilderImpl.deadLetterPolicy then rejects anything below 1. + Whether a subscription type can act on the policy is a client concern, so the runtime does not + gate on it. """ def _instance(self, max_message_retries=None, dead_letter_topic=None): @@ -172,13 +177,12 @@ def _instance(self, max_message_retries=None, dead_letter_topic=None): def test_no_retry_details_means_no_policy(self): instance = self._instance() - self.assertIsNone( - instance.get_dead_letter_policy(pulsar._pulsar.ConsumerType.Shared)) + self.assertIsNone(instance.get_dead_letter_policy()) def test_policy_built_from_retry_details(self): instance = self._instance(max_message_retries=3, dead_letter_topic="persistent://public/default/my-dlq") - policy = instance.get_dead_letter_policy(pulsar._pulsar.ConsumerType.Shared) + policy = instance.get_dead_letter_policy() self.assertIsNotNone(policy) self.assertEqual(3, policy.max_redeliver_count) @@ -188,38 +192,33 @@ def test_empty_dead_letter_topic_defers_to_client_default(self): # The Java runtime only sets the topic when non-empty, leaving the client to derive # "--DLQ". Passing "" through would override that with an invalid name. instance = self._instance(max_message_retries=2) - policy = instance.get_dead_letter_policy(pulsar._pulsar.ConsumerType.Shared) + policy = instance.get_dead_letter_policy() self.assertIsNotNone(policy) self.assertEqual(2, policy.max_redeliver_count) - def test_zero_retries_attaches_no_policy(self): - # Java accepts maxMessageRetries >= 0, but ConsumerDeadLetterPolicy rejects a redelivery count - # below 1, so zero cannot be expressed here. It must not raise and take the instance down. + def test_zero_retries_fails_fast(self): + # Java does not start with this value either: PulsarSource forwards 0 and + # ConsumerBuilderImpl.deadLetterPolicy rejects "MaxRedeliverCount must be > 0". Returning None + # here instead would let localrun - which bypasses validateNonJavaFunction - start with retries + # silently disabled while Java fails. instance = self._instance(max_message_retries=0, dead_letter_topic="persistent://public/default/my-dlq") - self.assertIsNone( - instance.get_dead_letter_policy(pulsar._pulsar.ConsumerType.Shared)) + with self.assertRaises(ValueError): + instance.get_dead_letter_policy() - def test_negative_retries_attaches_no_policy(self): + def test_negative_retries_fails_fast(self): instance = self._instance(max_message_retries=-1) - self.assertIsNone( - instance.get_dead_letter_policy(pulsar._pulsar.ConsumerType.Shared)) - - def test_key_shared_subscription_gets_policy(self): - instance = self._instance(max_message_retries=3) - self.assertIsNotNone( - instance.get_dead_letter_policy(pulsar._pulsar.ConsumerType.KeyShared)) + with self.assertRaises(ValueError): + instance.get_dead_letter_policy() - def test_failover_subscription_gets_no_policy(self): - # A dead letter policy has no effect on Failover, which retainOrdering and EFFECTIVELY_ONCE - # both select. Returning None keeps that explicit rather than silently ineffective. + def test_policy_is_not_gated_on_subscription_type(self): + # Subscription-type support is a client concern; the Java runtime always forwards a configured + # policy. Gating here would add a second support matrix that can drift from the client. instance = self._instance(max_message_retries=3, dead_letter_topic="persistent://public/default/my-dlq") - self.assertIsNone( - instance.get_dead_letter_policy(pulsar._pulsar.ConsumerType.Failover)) + policy = instance.get_dead_letter_policy() + + self.assertIsNotNone(policy) + self.assertEqual(3, policy.max_redeliver_count) - def test_exclusive_subscription_gets_no_policy(self): - instance = self._instance(max_message_retries=3) - self.assertIsNone( - instance.get_dead_letter_policy(pulsar._pulsar.ConsumerType.Exclusive)) From 1a5e054acfc7d52ff5498ff601f2dcd7deec96c5 Mon Sep 17 00:00:00 2001 From: david-streamlio <35466513+david-streamlio@users.noreply.github.com> Date: Thu, 27 Aug 2026 08:10:42 -0700 Subject: [PATCH 6/6] [fix][fn] Remove duplicate Go retry test and cover the retry edges Merging master brought in #26421, which had independently added a test named testGoFunctionStillRejectsMessageRetries. Both copies landed in different regions of the file, so the merge was clean and the collision only surfaced as a compile error: method testGoFunctionStillRejectsMessageRetries() is already defined Keep upstream's copy, which sits with the other Go tests and builds its config from minimalGoFunctionConfig() rather than routing through a Python one, and carry the explanatory comment onto it. Also cover three retry edges that had no test: - a negative maxMessageRetries means "unset" on the Python path, as it does on the Java path, and convert() emits no retryDetails for it - a dead letter topic with maxMessageRetries unset is rejected by doCommonChecks; nothing asserted that message - doGolangChecks refuses every count >= 0, not only a positive one, unlike Python which refuses only 0 Verified locally under JDK 25: 48 tests, 0 failures. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_017DmEAdjzyB9hJthimd3TZf --- .../utils/FunctionConfigUtilsTest.java | 37 ++++++++++++++++--- 1 file changed, 32 insertions(+), 5 deletions(-) diff --git a/pulsar-functions/utils/src/test/java/org/apache/pulsar/functions/utils/FunctionConfigUtilsTest.java b/pulsar-functions/utils/src/test/java/org/apache/pulsar/functions/utils/FunctionConfigUtilsTest.java index 1611fb35e3589..9aded020f2cec 100644 --- a/pulsar-functions/utils/src/test/java/org/apache/pulsar/functions/utils/FunctionConfigUtilsTest.java +++ b/pulsar-functions/utils/src/test/java/org/apache/pulsar/functions/utils/FunctionConfigUtilsTest.java @@ -562,13 +562,28 @@ public void testPythonFunctionRejectsZeroMessageRetries() { FunctionConfigUtils.validateNonJavaFunction(functionConfig); } + @Test + public void testPythonFunctionTreatsNegativeMessageRetriesAsUnset() { + // doPythonChecks refuses exactly 0; a negative count means "unset", as it does on the Java path, + // and convert() emits retryDetails only for a count >= 0. Pinned so tightening that guard to <= 0 + // cannot pass silently. + FunctionConfig functionConfig = createPythonFunctionConfig(); + functionConfig.setMaxMessageRetries(-1); + + FunctionConfigUtils.validateNonJavaFunction(functionConfig); + + assertFalse(FunctionConfigUtils.convert(functionConfig).hasRetryDetails()); + } + @Test(expectedExceptions = IllegalArgumentException.class, - expectedExceptionsMessageRegExp = "Message retries not yet supported in Go function") - public void testGoFunctionStillRejectsMessageRetries() { - // The Go runtime does not honour retryDetails yet, so its guard stays until it does. + expectedExceptionsMessageRegExp = "Dead Letter Topic specified, however max retries is set to infinity") + public void testPythonFunctionRejectsDeadLetterTopicWithoutMessageRetries() { + // doCommonChecks refuses a dead letter topic nothing can route to: with maxMessageRetries unset the + // redelivery count is infinite, so the topic would never receive anything. Same shape as the zero + // case above, reached from the other side. FunctionConfig functionConfig = createPythonFunctionConfig(); - functionConfig.setRuntime(FunctionConfig.Runtime.GO); - functionConfig.setMaxMessageRetries(3); + functionConfig.setDeadLetterTopic("test-dlq"); + FunctionConfigUtils.validateNonJavaFunction(functionConfig); } @@ -868,9 +883,21 @@ public void testGoFunctionRejectsRetainKeyOrderingWithEffectivelyOnce() { @Test(expectedExceptions = IllegalArgumentException.class, expectedExceptionsMessageRegExp = "Message retries not yet supported in Go function") public void testGoFunctionStillRejectsMessageRetries() { + // The Go runtime does not honour retryDetails yet, so its guard stays until it does. FunctionConfig functionConfig = minimalGoFunctionConfig(); functionConfig.setMaxMessageRetries(3); FunctionConfigUtils.validateNonJavaFunction(functionConfig); } + + @Test(expectedExceptions = IllegalArgumentException.class, + expectedExceptionsMessageRegExp = "Message retries not yet supported in Go function") + public void testGoFunctionRejectsZeroMessageRetries() { + // doGolangChecks refuses every count >= 0, not only a positive one -- unlike Python, which refuses + // only 0. Pinned so the two guards cannot be quietly aligned. + FunctionConfig functionConfig = minimalGoFunctionConfig(); + functionConfig.setMaxMessageRetries(0); + + FunctionConfigUtils.validateNonJavaFunction(functionConfig); + } }