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/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 672134c0689cc..17a9f2a094e7b 100755 --- a/pulsar-functions/instance/src/main/python/python_instance.py +++ b/pulsar-functions/instance/src/main/python/python_instance.py @@ -140,90 +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 - - 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, - **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 - } - 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: @@ -582,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(). @@ -602,6 +616,31 @@ 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 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 + "--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 + 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) + 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..99e4925467710 100644 --- a/pulsar-functions/instance/src/test/python/test_python_instance.py +++ b/pulsar-functions/instance/src/test/python/test_python_instance.py @@ -396,3 +396,181 @@ 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 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. + + 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, 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) + + 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): + # 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 + # "--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 + # 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. + # + # 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/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..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,6 +534,64 @@ public void testMergeRuntimeFlags() { ); } + @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; + } + @SuppressWarnings("deprecation") private FunctionConfig createFunctionConfig() { FunctionConfig functionConfig = new FunctionConfig(); @@ -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); + } }