[fix][fn] Support deadLetterTopic and maxMessageRetries for Python functions

pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdFunctions.java

@@ -371,7 +371,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")
@@ -380,7 +380,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).")

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

@@ -156,6 +156,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()
+
     subscription_name = self.instance_config.function_details.source.subscriptionName
 
     if not (subscription_name and subscription_name.strip()):
@@ -184,6 +186,7 @@ def run(self):
         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
       )
 
@@ -208,7 +211,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
       }
       consumer_args.update(nack_args)
       if consumer_conf.HasField("receiverQueueSize"):
@@ -602,6 +606,30 @@ def get_negative_ack_args(self):
 
     return {"negative_ack_redelivery_delay_ms": delay_ms}
 
+  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
+
+    return pulsar.ConsumerDeadLetterPolicy(
+      max_redeliver_count=retry_details.maxMessageRetries,
+      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
@@ -396,3 +397,75 @@ def test_result_is_splattable_into_subscribe_kwargs(self):
     args = self._instance(delay_ms=250).get_negative_ack_args()
     self.assertIsInstance(args, dict)
     self.assertEqual(["negative_ack_redelivery_delay_ms"], list(args.keys()))
+
+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.
+
+  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):
+    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())
+
+  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()
+
+    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()
+
+    self.assertIsNotNone(policy)
+    self.assertEqual(2, policy.max_redeliver_count)
+
+  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")
+    with self.assertRaises(ValueError):
+      instance.get_dead_letter_policy()
+
+  def test_negative_retries_fails_fast(self):
+    instance = self._instance(max_message_retries=-1)
+    with self.assertRaises(ValueError):
+      instance.get_dead_letter_policy()
+
+  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)
+

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");
         }
     }
 

pulsar-functions/utils/src/test/java/org/apache/pulsar/functions/utils/FunctionConfigUtilsTest.java

@@ -535,6 +535,64 @@ 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
+    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 = "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.setDeadLetterTopic("test-dlq");
+
+        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");
@@ -825,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);
+    }
 }