[fix][fn] Honour deadLetterTopic and maxMessageRetries in the Python function runtime

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:

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