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/7] [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
"<topic>-<subscription>-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 "<topic>-<subscription>-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
+    # "<topic>-<subscription>-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/7] [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 "<topic>-<subscription>-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/7] [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/7] [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/7] [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 "<topic>-<subscription>-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
     # "<topic>-<subscription>-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/7] [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) <noreply@anthropic.com>
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);
+    }
 }

From 94b4fcdf932ec4818ceacbe52f9b6c4019330760 Mon Sep 17 00:00:00 2001
From: david-streamlio <35466513+david-streamlio@users.noreply.github.com>
Date: Tue, 1 Sep 2026 18:16:02 -0400
Subject: [PATCH 7/7] [fix][fn] Address review on the Python dead letter policy

Motivation:

Review feedback on #26400 raised six points: the runtime's minimum
pulsar-client-python is now 3.3.0 and is undocumented, nothing exercises
either subscribe() call site, two tests do not pin what they claim to,
and two smaller slips.

Modifications:

- Document the client requirement. Client.subscribe() grew its
  dead_letter_policy parameter in 3.3.0, and the runtime passes the
  keyword on every subscription, so 3.3.0 is the floor. Every other
  argument the runtime passes is already present in 3.2.0, so this is
  the only thing setting it. README.md now states the requirement, notes
  that it is only visible on a self-managed worker, and shows where to
  pin it for a zip-packaged function.

- Extract setup_consumers() from run(), giving the consumer path the
  same testable seam setup_producer() already has, and cover both
  subscribe() call sites - the topicsToSerDeClassName path, the
  inputSpecs path and its regex variant - with assertions on the kwargs
  a mocked client receives.

- Drop the "or None" on deadLetterTopic. It could not change behaviour:
  ConsumerDeadLetterPolicy skips the builder call only for None, and
  DeadLetterPolicyBuilder.deadLetterTopic("") leaves getDeadLetterTopic()
  == "", which is what an unset policy also reports. The test that
  claimed to pin it now pins the runtime's actual half of the contract,
  and its comment no longer misdescribes the client.

- Make test_policy_is_not_gated_on_subscription_type exercise the
  subscription types the earlier gate suppressed. It used a default
  instance, which selects Shared - the one case the gate allowed - so it
  passed with the gate fully restored.

- Remove a duplicate "import pulsar" and move
  @SuppressWarnings("deprecation") back onto createFunctionConfig(), the
  method it was suppressing for.
---
 .../instance/src/main/python/README.md        |  20 ++
 .../src/main/python/python_instance.py        | 193 +++++++++---------
 .../src/test/python/test_python_instance.py   | 125 +++++++++++-
 .../utils/FunctionConfigUtilsTest.java        |   2 +-
 4 files changed, 238 insertions(+), 102 deletions(-)

diff --git a/pulsar-functions/instance/src/main/python/README.md b/pulsar-functions/instance/src/main/python/README.md
index 6465d88513115..1550ba2de68a8 100644
--- a/pulsar-functions/instance/src/main/python/README.md
+++ b/pulsar-functions/instance/src/main/python/README.md
@@ -1,5 +1,25 @@
 # Pulsar Functions Python Runtime
 
+### pulsar-client-python requirements
+
+The runtime requires **pulsar-client-python 3.3.0 or newer**. `Client.subscribe()` grew its
+`dead_letter_policy` parameter in 3.3.0 (as did `ConsumerDeadLetterPolicy`, in the same release), and
+the runtime passes the keyword on every subscription — as `None` when the function configures no
+`maxMessageRetries` or `deadLetterTopic`, which `Client.subscribe()` treats as a no-op. Every other
+argument the runtime passes is present in 3.2.0, so this is the only thing setting the floor.
+
+Everything Pulsar ships already satisfies this: the `pulsar-client-python` version in
+[`gradle/libs.versions.toml`](../../../../../gradle/libs.versions.toml) is what the
+[Docker images](../../../../../docker/pulsar/Dockerfile) install and what CI runs the instance tests
+against. The floor is only visible on a self-managed worker, where the process runtime launches the
+host's `python3` (see `RuntimeUtils`) and the installed client is the operator's. For a
+zip-packaged function, pin it in the function's own `requirements.txt`, which
+`python_instance_main.py` pip-installs before the instance starts:
+
+```
+pulsar-client>=3.3.0
+```
+
 ### Producer configuration
 
 Both producers the runtime creates — the sink (output topic) producer in `python_instance.py` and the
diff --git a/pulsar-functions/instance/src/main/python/python_instance.py b/pulsar-functions/instance/src/main/python/python_instance.py
index f30b30aaf3014..17a9f2a094e7b 100755
--- a/pulsar-functions/instance/src/main/python/python_instance.py
+++ b/pulsar-functions/instance/src/main/python/python_instance.py
@@ -140,94 +140,7 @@ def run(self):
     self.state_context = self.setup_state()
 
     # Setup consumers and input deserializers
-    mode = pulsar._pulsar.ConsumerType.Shared
-    if self.instance_config.function_details.source.subscriptionType == Function_pb2.SubscriptionType.Value("FAILOVER"):
-      mode = pulsar._pulsar.ConsumerType.Failover
-
-    if self.instance_config.function_details.retainOrdering or \
-      self.instance_config.function_details.processingGuarantees == Function_pb2.ProcessingGuarantees.Value("EFFECTIVELY_ONCE"):
-      mode = pulsar._pulsar.ConsumerType.Failover
-    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
-
-    dead_letter_policy = self.get_dead_letter_policy()
-
-    subscription_name = self.instance_config.function_details.source.subscriptionName
-
-    if not (subscription_name and subscription_name.strip()):
-      subscription_name = str(self.instance_config.function_details.tenant) + "/" + \
-                          str(self.instance_config.function_details.namespace) + "/" + \
-                          str(self.instance_config.function_details.name)
-
-    properties = util.get_properties(util.getFullyQualifiedFunctionName(
-                        self.instance_config.function_details.tenant,
-                        self.instance_config.function_details.namespace,
-                        self.instance_config.function_details.name),
-                        self.instance_config.instance_id)
-
-    for topic, serde in self.instance_config.function_details.source.topicsToSerDeClassName.items():
-      if not serde:
-        serde_kclass = util.import_class(os.path.dirname(self.user_code), DEFAULT_SERIALIZER)
-      else:
-        serde_kclass = util.import_class(os.path.dirname(self.user_code), serde)
-      self.input_serdes[topic] = serde_kclass()
-      Log.debug("Setting up consumer for topic %s with subname %s" % (topic, subscription_name))
-
-      self.consumers[topic] = self.pulsar_client.subscribe(
-        str(topic), subscription_name,
-        consumer_type=mode,
-        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,
-        dead_letter_policy=dead_letter_policy,
-        **nack_args
-      )
-
-    for topic, consumer_conf in self.instance_config.function_details.source.inputSpecs.items():
-      if not consumer_conf.serdeClassName:
-        serde_kclass = util.import_class(os.path.dirname(self.user_code), DEFAULT_SERIALIZER)
-      else:
-        serde_kclass = util.import_class(os.path.dirname(self.user_code), consumer_conf.serdeClassName)
-      self.input_serdes[topic] = serde_kclass()
-
-      self.input_schema[topic] = self.get_schema(consumer_conf.schemaType,
-                                                 self.instance_config.function_details.source.typeClassName,
-                                                 consumer_conf.schemaProperties)
-      Log.debug("Setting up consumer for topic %s with subname %s" % (topic, subscription_name))
-
-      crypto_key_reader = self.get_crypto_reader(consumer_conf.cryptoSpec)
-
-      consumer_args = {
-        "consumer_type": mode,
-        "schema": self.input_schema[topic],
-        "message_listener": partial(self.message_listener, self.input_serdes[topic], self.input_schema[topic]),
-        "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,
-        "dead_letter_policy": dead_letter_policy
-      }
-      consumer_args.update(nack_args)
-      if consumer_conf.HasField("receiverQueueSize"):
-        consumer_args["receiver_queue_size"] = consumer_conf.receiverQueueSize.value
-
-      if consumer_conf.isRegexPattern:
-        self.consumers[topic] = self.pulsar_client.subscribe(
-          re.compile(str(topic)), subscription_name,
-          **consumer_args
-        )
-      else:
-        self.consumers[topic] = self.pulsar_client.subscribe(
-          str(topic), subscription_name,
-          **consumer_args
-        )
+    self.setup_consumers()
 
     function_kclass = util.import_class(os.path.dirname(self.user_code), self.instance_config.function_details.className)
     if function_kclass is None:
@@ -586,6 +499,103 @@ def get_record_class(self, class_name):
         except:
           pass
       return record_kclass
+  def setup_consumers(self):
+    """Subscribe to every input topic and build the matching input deserializers.
+
+    Kept separate from run() so the consumer configuration can be exercised the same way
+    setup_producer() is: the two subscribe() call sites below - the topicsToSerDeClassName path
+    and the inputSpecs path, which also covers the regex variant - must stay in step with each
+    other, and only a test that reaches the client can show that they do.
+    """
+    mode = pulsar._pulsar.ConsumerType.Shared
+    if self.instance_config.function_details.source.subscriptionType == Function_pb2.SubscriptionType.Value("FAILOVER"):
+      mode = pulsar._pulsar.ConsumerType.Failover
+
+    if self.instance_config.function_details.retainOrdering or \
+      self.instance_config.function_details.processingGuarantees == Function_pb2.ProcessingGuarantees.Value("EFFECTIVELY_ONCE"):
+      mode = pulsar._pulsar.ConsumerType.Failover
+    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
+
+    dead_letter_policy = self.get_dead_letter_policy()
+
+    subscription_name = self.instance_config.function_details.source.subscriptionName
+
+    if not (subscription_name and subscription_name.strip()):
+      subscription_name = str(self.instance_config.function_details.tenant) + "/" + \
+                          str(self.instance_config.function_details.namespace) + "/" + \
+                          str(self.instance_config.function_details.name)
+
+    properties = util.get_properties(util.getFullyQualifiedFunctionName(
+                        self.instance_config.function_details.tenant,
+                        self.instance_config.function_details.namespace,
+                        self.instance_config.function_details.name),
+                        self.instance_config.instance_id)
+
+    for topic, serde in self.instance_config.function_details.source.topicsToSerDeClassName.items():
+      if not serde:
+        serde_kclass = util.import_class(os.path.dirname(self.user_code), DEFAULT_SERIALIZER)
+      else:
+        serde_kclass = util.import_class(os.path.dirname(self.user_code), serde)
+      self.input_serdes[topic] = serde_kclass()
+      Log.debug("Setting up consumer for topic %s with subname %s" % (topic, subscription_name))
+
+      self.consumers[topic] = self.pulsar_client.subscribe(
+        str(topic), subscription_name,
+        consumer_type=mode,
+        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,
+        dead_letter_policy=dead_letter_policy,
+        **nack_args
+      )
+
+    for topic, consumer_conf in self.instance_config.function_details.source.inputSpecs.items():
+      if not consumer_conf.serdeClassName:
+        serde_kclass = util.import_class(os.path.dirname(self.user_code), DEFAULT_SERIALIZER)
+      else:
+        serde_kclass = util.import_class(os.path.dirname(self.user_code), consumer_conf.serdeClassName)
+      self.input_serdes[topic] = serde_kclass()
+
+      self.input_schema[topic] = self.get_schema(consumer_conf.schemaType,
+                                                 self.instance_config.function_details.source.typeClassName,
+                                                 consumer_conf.schemaProperties)
+      Log.debug("Setting up consumer for topic %s with subname %s" % (topic, subscription_name))
+
+      crypto_key_reader = self.get_crypto_reader(consumer_conf.cryptoSpec)
+
+      consumer_args = {
+        "consumer_type": mode,
+        "schema": self.input_schema[topic],
+        "message_listener": partial(self.message_listener, self.input_serdes[topic], self.input_schema[topic]),
+        "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,
+        "dead_letter_policy": dead_letter_policy
+      }
+      consumer_args.update(nack_args)
+      if consumer_conf.HasField("receiverQueueSize"):
+        consumer_args["receiver_queue_size"] = consumer_conf.receiverQueueSize.value
+
+      if consumer_conf.isRegexPattern:
+        self.consumers[topic] = self.pulsar_client.subscribe(
+          re.compile(str(topic)), subscription_name,
+          **consumer_args
+        )
+      else:
+        self.consumers[topic] = self.pulsar_client.subscribe(
+          str(topic), subscription_name,
+          **consumer_args
+        )
+
   def get_negative_ack_args(self):
     """Build the negative-ack redelivery delay argument for Client.subscribe().
 
@@ -610,8 +620,9 @@ 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 "<topic>-<subscription>-DLQ".
+    when retryDetails is present, and the deadLetterTopic is forwarded as configured. An unset one
+    reads as "" from the proto, and the client treats an empty topic as unset - it derives
+    "<topic>-<subscription>-DLQ" from the topic and subscription instead.
 
     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
@@ -628,7 +639,7 @@ def get_dead_letter_policy(self):
 
     return pulsar.ConsumerDeadLetterPolicy(
       max_redeliver_count=retry_details.maxMessageRetries,
-      dead_letter_topic=retry_details.deadLetterTopic or None)
+      dead_letter_topic=retry_details.deadLetterTopic)
 
   def get_crypto_reader(self, crypto_spec):
     crypto_key_reader = 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 445622ae71233..99e4925467710 100644
--- a/pulsar-functions/instance/src/test/python/test_python_instance.py
+++ b/pulsar-functions/instance/src/test/python/test_python_instance.py
@@ -32,7 +32,6 @@
 
 from contextimpl import ContextImpl
 from python_instance import PythonInstance, InstanceConfig
-import pulsar
 from pulsar import Message
 
 import Function_pb2
@@ -398,6 +397,77 @@ def test_result_is_splattable_into_subscribe_kwargs(self):
     self.assertIsInstance(args, dict)
     self.assertEqual(["negative_ack_redelivery_delay_ms"], list(args.keys()))
 
+class TestConsumerSubscribeArgs(unittest.TestCase):
+  """The dead letter policy must actually reach Client.subscribe().
+
+  get_dead_letter_policy() returning the right object is only half of it: setup_consumers()
+  subscribes on two paths - source.topicsToSerDeClassName and source.inputSpecs, the latter also
+  covering the regex variant - and the policy has to reach the client on each. Without this, the
+  policy could be dropped from one call site, or passed under a keyword the client does not accept,
+  and the rest of the suite would stay green.
+  """
+
+  def _subscribe_kwargs(self, max_message_retries=None, dead_letter_topic=None,
+                        via_input_specs=False, regex=False):
+    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
+
+    topic = "persistent://public/default/.*" if regex else "persistent://public/default/in"
+    if via_input_specs:
+      consumer_conf = function_details.source.inputSpecs[topic]
+      consumer_conf.isRegexPattern = regex
+    else:
+      function_details.source.topicsToSerDeClassName[topic] = ""
+
+    mock_pulsar_client = Mock()
+    instance = PythonInstance('test_instance', 'test_func', '1.0', function_details, 100, 30,
+                              'user_code', mock_pulsar_client, Mock(), 'test_cluster', 'test_url',
+                              None)
+    instance.get_schema = Mock(return_value="DEFAULT_SCHEMA")
+    instance.get_crypto_reader = Mock(return_value=None)
+    instance.setup_consumers()
+
+    self.assertEqual(1, mock_pulsar_client.subscribe.call_count)
+    _, kwargs = mock_pulsar_client.subscribe.call_args
+    return kwargs
+
+  def test_policy_reaches_the_topics_to_serde_path(self):
+    kwargs = self._subscribe_kwargs(max_message_retries=3,
+                                    dead_letter_topic="persistent://public/default/my-dlq")
+
+    policy = kwargs["dead_letter_policy"]
+    self.assertEqual(3, policy.max_redeliver_count)
+    self.assertEqual("persistent://public/default/my-dlq", policy.dead_letter_topic)
+
+  def test_policy_reaches_the_input_specs_path(self):
+    kwargs = self._subscribe_kwargs(max_message_retries=3,
+                                    dead_letter_topic="persistent://public/default/my-dlq",
+                                    via_input_specs=True)
+
+    policy = kwargs["dead_letter_policy"]
+    self.assertEqual(3, policy.max_redeliver_count)
+    self.assertEqual("persistent://public/default/my-dlq", policy.dead_letter_topic)
+
+  def test_policy_reaches_the_regex_path(self):
+    kwargs = self._subscribe_kwargs(max_message_retries=3,
+                                    dead_letter_topic="persistent://public/default/my-dlq",
+                                    via_input_specs=True, regex=True)
+
+    self.assertEqual(3, kwargs["dead_letter_policy"].max_redeliver_count)
+
+  def test_keyword_is_passed_as_none_without_retry_details(self):
+    # The keyword is always passed, so the runtime requires pulsar-client-python 3.3.0 or newer -
+    # see README.md. Client.subscribe() guards with "if dead_letter_policy:", so None is a no-op
+    # there. Both paths must agree on this; one of them dropping the keyword would be the kind of
+    # drift this class exists to catch.
+    self.assertIsNone(self._subscribe_kwargs()["dead_letter_policy"])
+    self.assertIsNone(self._subscribe_kwargs(via_input_specs=True)["dead_letter_policy"])
+
+
 class TestDeadLetterPolicy(unittest.TestCase):
   """Covers FunctionDetails.retryDetails -> ConsumerDeadLetterPolicy.
 
@@ -411,13 +481,15 @@ class TestDeadLetterPolicy(unittest.TestCase):
   gate on it.
   """
 
-  def _instance(self, max_message_retries=None, dead_letter_topic=None):
+  def _instance(self, max_message_retries=None, dead_letter_topic=None, configure=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
+    if configure is not None:
+      configure(function_details)
 
     return PythonInstance('test_instance', 'test_func', '1.0', function_details, 100, 30,
                           'user_code', Mock(), Mock(), 'test_cluster', 'test_url', None)
@@ -436,13 +508,21 @@ def test_policy_built_from_retry_details(self):
     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
-    # "<topic>-<subscription>-DLQ". Passing "" through would override that with an invalid name.
+    # An unset deadLetterTopic reads as "" from the proto and is forwarded as configured. The
+    # client treats an empty topic as unset: ConsumerDeadLetterPolicy skips the builder call only
+    # for None, and DeadLetterPolicyBuilder.deadLetterTopic("") leaves getDeadLetterTopic() == "",
+    # which is exactly what an unset policy reports. Either way the client derives
+    # "<topic>-<subscription>-DLQ".
+    #
+    # So what is pinned here is the runtime's half of that: retryDetails without a topic still
+    # produces a policy carrying the redelivery count, and the runtime does not invent a DLQ name
+    # of its own.
     instance = self._instance(max_message_retries=2)
     policy = instance.get_dead_letter_policy()
 
     self.assertIsNotNone(policy)
     self.assertEqual(2, policy.max_redeliver_count)
+    self.assertEqual("", policy.dead_letter_topic)
 
   def test_zero_retries_fails_fast(self):
     # Java does not start with this value either: PulsarSource forwards 0 and
@@ -462,10 +542,35 @@ def test_negative_retries_fails_fast(self):
   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")
-    policy = instance.get_dead_letter_policy()
-
-    self.assertIsNotNone(policy)
-    self.assertEqual(3, policy.max_redeliver_count)
+    #
+    # Each case below is one the earlier revision's gate suppressed the policy for. run() derives
+    # the consumer type from FunctionDetails - retainOrdering, EFFECTIVELY_ONCE and a FAILOVER
+    # subscriptionType each select Failover, and retainKeyOrdering selects KeyShared - so a gate
+    # reintroduced here would have to read one of these, and would fail this test. A default
+    # instance selects Shared, which the gate allowed, so it cannot pin this on its own.
+    def retain_ordering(details):
+      details.retainOrdering = True
+
+    def effectively_once(details):
+      details.processingGuarantees = Function_pb2.ProcessingGuarantees.Value("EFFECTIVELY_ONCE")
+
+    def failover_subscription(details):
+      details.source.subscriptionType = Function_pb2.SubscriptionType.Value("FAILOVER")
+
+    def retain_key_ordering(details):
+      details.retainKeyOrdering = True
+
+    for label, configure in (("retainOrdering", retain_ordering),
+                             ("EFFECTIVELY_ONCE", effectively_once),
+                             ("FAILOVER subscriptionType", failover_subscription),
+                             ("retainKeyOrdering", retain_key_ordering)):
+      with self.subTest(selects_consumer_type_via=label):
+        instance = self._instance(max_message_retries=3,
+                                  dead_letter_topic="persistent://public/default/my-dlq",
+                                  configure=configure)
+        policy = instance.get_dead_letter_policy()
+
+        self.assertIsNotNone(policy)
+        self.assertEqual(3, policy.max_redeliver_count)
+        self.assertEqual("persistent://public/default/my-dlq", policy.dead_letter_topic)
 
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 9aded020f2cec..91effaa9244bd 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
@@ -534,7 +534,6 @@ public void testMergeRuntimeFlags() {
         );
     }
 
-    @SuppressWarnings("deprecation")
     @Test
     public void testPythonFunctionAcceptsMessageRetriesAndDeadLetterTopic() {
         FunctionConfig functionConfig = createPythonFunctionConfig();
@@ -593,6 +592,7 @@ private FunctionConfig createPythonFunctionConfig() {
         return functionConfig;
     }
 
+    @SuppressWarnings("deprecation")
     private FunctionConfig createFunctionConfig() {
         FunctionConfig functionConfig = new FunctionConfig();
         functionConfig.setTenant("test-tenant");