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 889d90e76e95d..8e82da8a94d11 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 @@ -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).") diff --git a/pulsar-functions/instance/src/main/python/python_instance.py b/pulsar-functions/instance/src/main/python/python_instance.py index 672134c0689cc..f30b30aaf3014 100755 --- a/pulsar-functions/instance/src/main/python/python_instance.py +++ b/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 "--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: 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 8667f13964b04..445622ae71233 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 @@ -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 + # "--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) + 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 81a3f50508cdd..3546a16f51e99 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 dd0eb8d98b709..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 @@ -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 "--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); + } }