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

[david-streamlio]

Fixes #26397

Motivation

FunctionConfig accepts maxMessageRetries and deadLetterTopic, both are carried into the instance as FunctionDetails.retryDetails (Function.proto L58-61, L91), and the Java runtime applies them. The Python runtime ignored them entirely — grep -i "dead_letter\|retryDetails" python_instance.py returned nothing on master.

The failure mode is silent, which is the damaging part. This is accepted without warning:

pulsar-admin functions create --py fn.py --classname fn.F \
  --dead-letter-topic persistent://public/default/my-dlq \
  --max-message-retries 3 ...

functions get reports the configuration back faithfully, and at runtime nothing is ever routed to the DLQ.

Teaching the runtime to honour retryDetails is necessary but not sufficient. Two gates in FunctionConfigUtils line up exactly and, between them, keep a Python function from ever reaching a retryDetails message on the cluster path:

• doPythonChecks refuses any maxMessageRetries >= 0 outright ("Message retries not yet supported in python").
• convert only populates retryDetails when maxMessageRetries != null && >= 0 (L304) — the very condition doPythonChecks rejects. So --dead-letter-topic on its own never produces a retryDetails message at all, and the runtime has nothing to honour.

validateNonJavaFunction has a single caller, the worker REST API (FunctionsImpl), so this refusal applies to cluster submission only. LocalRunner never calls it, which is why the runtime path is reachable under localrun today and nowhere else. Fixing only the runtime would ship a fix that no user could invoke.

Modifications

Runtime (python_instance.py). Add get_dead_letter_policy() to PythonInstance and pass its result to all three subscribe() call sites in run() — the topicsToSerDeClassName loop and both branches of the inputSpecs loop.

The rules follow the Java runtime:

• Guard on HasField("retryDetails"), matching JavaInstanceRunnable's hasRetryDetails() check. HasField is already the idiom in this file (used for receiverQueueSize).
• Set the dead letter topic only when non-empty, matching PulsarSource, so the client derives its <topic>-<subscription>-DLQ default rather than receiving an empty name.

Validation (FunctionConfigUtils.doPythonChecks). Replace the blanket refusal 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 (pulsar/__init__.py L761-762). Accepting it would create a function whose dead letter topic never receives anything — precisely the silently ineffective configuration this change 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.

CLI metadata (CmdFunctions). The @Option descriptions carry a runtime marker that the docs sync parses into the Support column of the published pulsar-admin CLI reference, and that pulsar-admin functions create --help prints verbatim. --max-message-retries and --dead-letter-topic were both marked #Java; they are now #Java, Python.

doGolangChecks keeps its guard: the Go runtime does not honour retryDetails yet, and a test now pins that so this cannot be widened to Go by accident.

The pre-existing cross-cutting checks in doCommonChecks still apply to Python and cover the rest of the space: a dead letter topic with retries unset is refused ("Dead Letter Topic specified, however max retries is set to infinity"), as is maxMessageRetries combined with EFFECTIVELY_ONCE.

Two runtime cases still cannot mirror Java exactly, and both warn rather than failing the instance or silently doing nothing:

1. maxMessageRetries == 0. Now rejected at creation on the cluster path, but still reachable under localrun, which skips validateNonJavaFunction. Attaching no policy is the only available behaviour there; the warning names the dead letter topic that will not receive messages. Raising would take down a function the Java runtime would have started.
2. Non-Shared subscriptions. A dead letter policy only takes effect on Shared and KeyShared. retainOrdering and EFFECTIVELY_ONCE both select Failover, where the policy would be silently ineffective — the same class of bug as this issue — so that combination warns as well.

I did not change the Java-side >= 0 behaviour or the client's >= 1 constraint; reconciling them is a larger discussion than this fix.

Verifying this change

• Make sure that the change passes the CI checks.

This change added tests and can be verified as follows:

• 8 unit tests in test_python_instance.py (TestDeadLetterPolicy), covering: no retryDetails → no policy; a policy built from retryDetails; an empty dead letter topic deferring to the client default; maxMessageRetries of 0 and of -1 attaching no policy rather than raising; KeyShared receiving a policy; Failover and Exclusive not.
• 4 unit tests in FunctionConfigUtilsTest.java: Python accepts retries with and without an explicit dead letter topic, rejects zero with the new message, and Go still refuses retries.
• Full Python file: 12/12 pass via pulsar-functions/instance/src/scripts/run_python_instance_tests.sh. FunctionConfigUtilsTest: 41/41 pass.
• Confirmed the Python tests are not vacuous: making get_dead_letter_policy() return None unconditionally fails 3 of them.

Does this pull request potentially affect one of the following parts:

• Dependencies (add or upgrade a dependency)
• The public API
• The schema
• The default values of configurations
• The threading model
• The binary protocol
• The REST endpoints

A function that does not set retryDetails is unaffected: get_dead_letter_policy() returns None and dead_letter_policy=None is what the client already defaults to.

The validation change is relaxing only: configurations that were accepted before are still accepted. The one newly-rejected input, maxMessageRetries == 0 on a Python function, was previously rejected too, by the broader guard this replaces — no configuration that used to be creatable stops being creatable.

Documentation

• doc-required

docs/functions-cli.md in apache/pulsar-site documents maxMessageRetries and deadLetterTopic with no runtime qualification, so the table needs a note that Java and Python honour them (Python requiring at least 1) while Go does not. That table is hand-maintained, so it needs its own PR there; the #Java, Python marker above covers the generated CLI reference from this side.

For the record, the note added when #6084 was closed in 2020 -- "This parameter is not supported in Python Functions" -- is no longer present in the current docs. It survives only in versioned_docs/version-2.3.0 through version-2.10.x, where it was accurate for those releases and should stay; it was dropped from the live docs in the 2.11 functions-cli.md rewrite. So there is no incorrect statement to retract, only a missing caveat to add.

I have not opened the apache/pulsar-site PR yet -- flagging it so it is not lost, and happy to open it once the behaviour here is settled in review.

[david-streamlio]

A gap in this PR's reasoning that I found while auditing windowing support, and that a reviewer should weigh before merging.

This PR makes the Python runtime honour retryDetails, but the broker still refuses the configuration that would reach it:

// FunctionConfigUtils.doPythonChecks()
if (functionConfig.getMaxMessageRetries() != null && functionConfig.getMaxMessageRetries() >= 0) {
    throw new IllegalArgumentException("Message retries not yet supported in python");
}

So on master, pulsar-admin functions create --py ... --max-message-retries 3 fails at creation rather than being silently ignored at runtime. That makes the framing in #26397 and in this description partly wrong: I described the failure as silent, and for deadLetterTopic alone it is — that field has no such check, so it is accepted and dropped. But the pairing an operator would actually configure, --max-message-retries with --dead-letter-topic, is refused up front.

That changes what this PR is worth on its own:

• As it stands, it makes the runtime ready but the path is still closed. A user cannot exercise it without also relaxing doPythonChecks.
• Relaxing that check is a deliberate, separate decision — it is the broker declaring Python DLQ supported — and it should probably not be smuggled in through a runtime PR.

Options, and I do not have a strong view on which is right:

1. Merge this as-is and follow up with a small PR removing the doPythonChecks guard, so the runtime support demonstrably exists before the config surface opens.
2. Extend this PR to remove the guard too, so the feature is usable when it lands.
3. Hold it until there is a decision on whether Python DLQ is a supported feature at all.

I would lean toward 1, since it keeps the runtime change reviewable on its own merits and makes the enabling change an explicit, visible decision rather than a side effect. Happy to open the follow-up either way.

The same relationship exists for the Go runtime: doGolangChecks carries an identical maxMessageRetries guard, so #26406 will need the same consideration when it is picked up. negativeAckRedeliveryDelayMs (#26413, #26415) is unaffected — no equivalent check exists for it.

[david-streamlio]

Cross-linking for reviewers: the documentation for this change is apache/pulsar-site#1214 ("Note runtime support for maxMessageRetries and deadLetterTopic"), which is open and awaiting review.

The doc-required label here is correct and will be satisfied once that merges — leaving it in place deliberately, unlike #26392 and #26393 where the corresponding doc has already landed.

[freeznet]

The positive retry mapping and default DLQ topic handling look sound, but the two guards below introduce behavior and ownership differences from the Java runtime. Please align these paths before merging.

[freeznet]

Java does not start successfully with this value: PulsarSource forwards 0 to ConsumerBuilder.deadLetterPolicy(), and ConsumerBuilderImpl rejects maxRedeliverCount <= 0. Since LocalRunner bypasses validateNonJavaFunction, returning None here makes Python localrun start with retries silently disabled while Java fails fast. Please let ConsumerDeadLetterPolicy reject zero (or move the validation to a path shared by cluster and localrun) instead of swallowing it.

[david-streamlio]

Agreed, and fixed in 47df72c. I verified the Java path rather than taking it on trust, and it is exactly as you describe:

• PulsarSource.java:97 guards on getMaxMessageRetries() != null && >= 0, so 0 is forwarded rather than skipped;
• ConsumerBuilderImpl.java:530 then does checkArgument(deadLetterPolicy.getMaxRedeliverCount() > 0, "MaxRedeliverCount must be > 0.");
• validateNonJavaFunction has exactly one non-test caller, FunctionsImpl.java:829, so LocalRunner does bypass it.

I took the first of your two options: the value is now passed straight through and ConsumerDeadLetterPolicy rejects it. Confirmed against the pinned client that it raises ValueError: max_redeliver_count must be greater than 0 for both 0 and negatives, so localrun now fails the instance the same way Java does instead of starting with retries quietly disabled.

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.

[freeznet]

Could we remove this runtime-level subscription-type gate and the consumer_type parameter? The Java runtime always forwards a configured DeadLetterPolicy; Shared/KeyShared support is a client concern. Encoding the current native-client limitation here makes Python diverge from Java and adds a second support matrix that can become stale. Passing the policy through is simpler; any validation or warning should live at the common config or client boundary.

[david-streamlio]

Agreed, removed in 47df72c along with the consumer_type parameter; the caller is now get_dead_letter_policy().

Your reasoning is the part I found persuasive: the Java runtime always forwards a configured DeadLetterPolicy, so gating here created a second support matrix in the runtime that would drift from the client as native-client support changes. Passing the policy through is both simpler and one less thing to keep in sync.

The three subscription-type test cases are replaced by a single one asserting the policy is not gated on subscription type, so the intent is pinned rather than just untested.

[david-streamlio]

@freeznet both addressed in 47df72c — replies inline.

• The zero guard is gone; the value is passed through and ConsumerDeadLetterPolicy rejects it, so localrun fails the same way Java does.
• The subscription-type gate and the consumer_type parameter are gone; the policy is forwarded unconditionally, as in the Java runtime.

I checked PulsarSource, ConsumerBuilderImpl, and the validateNonJavaFunction call sites directly rather than taking the description on trust, and each matched. Tests reworked accordingly: 10 pass under the same invocation run_python_instance_tests.sh uses.

Both changes made the PR smaller, which is a good sign — thanks for pushing on the consistency argument.

[david-streamlio]

@freeznet both points are addressed in 47df72c:

• Zero retries — the value now passes straight through, so ConsumerDeadLetterPolicy rejects it (ValueError: max_redeliver_count must be greater than 0) for both 0 and negatives. Localrun now fails the instance the same way Java does instead of starting with retries quietly disabled. The cluster path keeps its earlier, friendlier diagnostic, since validateNonJavaFunction still rejects maxMessageRetries == 0 for Python before submission.
• Subscription-type gate — removed, along with the consumer_type parameter; the caller is now get_dead_letter_policy(). The three subscription-type test cases are replaced by a single one asserting the policy is not gated on subscription type, so the intent is pinned rather than just untested.

Since then I also pushed 1a5e054acf, which removes a duplicate testGoFunctionStillRejectsMessageRetries that arrived when master brought in #26421 — both copies landed in different regions of the file, so the merge was clean and it only surfaced as a compile error. That commit also adds coverage for three retry edges that had none: a negative maxMessageRetries meaning "unset" on the Python path, a dead letter topic with retries unset being rejected by doCommonChecks, and doGolangChecks refusing every count >= 0 rather than only a positive one.

CI is green on 1a5e054acf — 46/46 checks passing.

Could you take another look when you have a moment?

[lhotari]

The direction here is right and the two points @freeznet raised are genuinely fixed in 47df72c — I checked both rather than taking the replies on trust. Passing maxMessageRetries through so ConsumerDeadLetterPolicy rejects it matches PulsarSource.java:98-104 (>= 0 builds the policy, ConsumerBuilderImpl then rejects <= 0), and ConsumerDeadLetterPolicy.__init__ does raise ValueError: max_redeliver_count must be greater than 0 for both 0 and negatives. Removing the subscription-type gate is also the right call for the reason given: the Java runtime forwards a configured policy unconditionally.

What I do not think holds up is the claim, made in reply to the second thread, that the replacement test leaves the intent "pinned rather than just untested". I re-applied the exact gate that 5cc6207b061 introduced to a copy of get_dead_letter_policy() (deriving the consumer mode from function_details the way run() does) and all 33 tests in test_python_instance still passed. A positive control (get_dead_letter_policy() always returning None) fails 5 tests, so the harness works — the gate simply is not covered. The same is true of the or None on deadLetterTopic: dropping it leaves the suite green.

One thing worth writing down rather than coding around. dead_letter_policy is now passed to subscribe() on both consumer paths unconditionally, and that keyword only exists from pulsar-client-python 3.3.0, so this raises the Python runtime's effective client floor. The images we ship are already fine: docker/pulsar/Dockerfile:187 and docker/pulsar/Dockerfile.wolfi:111 install pulsar-client[all] at the version gradle/libs.versions.toml:22 pins (3.13.0), and run_python_instance_tests.sh:34 pins the same for CI. What is missing is that we never state the requirement anywhere — the process runtime launches the host's python3 (RuntimeUtils.java:420), so a self-managed worker runs whatever client the operator happens to have installed. A short note telling users to run a recent pulsar-client-python (>= 3.3.0) in pulsar-functions/instance/src/main/python/README.md, and in the pulsar-site page you already have open (apache/pulsar-site#1214), would close it.

The rest is small: a @SuppressWarnings("deprecation") that the insertion point silently moved off createFunctionConfig(), a duplicated import pulsar, and the fact that neither modified subscribe() call is exercised by any test.

On the broader question raised in the first PR comment (merge as-is vs. relax doPythonChecks here vs. hold): the PR as it stands has taken option 2, and I think that is the right choice — shipping the runtime support without opening the config surface would leave it unreachable dead code.

[lhotari]

[QUALITY] dead_letter_policy raises the Python runtime's minimum pulsar-client-python to 3.3.0 — worth stating in the requirements and docs

Both subscribe paths now always pass the keyword — here and again at line 215 — even when get_dead_letter_policy() returned None.

Client.subscribe() only grew a dead_letter_policy parameter in pulsar-client-python 3.3.0; it is absent from the v3.2.0 signature. Passing None is fine on 3.3.0+ (3.13.0 declares dead_letter_policy: Union[None, ConsumerDeadLetterPolicy] = None and guards with if dead_letter_policy:), so the practical effect is simply that the Python runtime now needs a 3.3.0-or-newer client.

Everything apache/pulsar ships already satisfies that: gradle/libs.versions.toml:22 pins 3.13.0, docker/pulsar/Dockerfile:187 and docker/pulsar/Dockerfile.wolfi:111 install it into the images, and run_python_instance_tests.sh:34 pins the same version for CI. The floor is only visible on a self-managed worker, where the process runtime launches the host's interpreter — RuntimeUtils.java:420 adds a bare "python3" — and the distribution ships the instance sources rather than a pinned wheel, so the installed client is the operator's.

So rather than defending the call against old clients, I would state the requirement. The Python runtime's dependency documentation should instruct users to run a recent pulsar-client-python (>= 3.3.0): a line in pulsar-functions/instance/src/main/python/README.md, the matching note in the pulsar-site page you already opened (apache/pulsar-site#1214), and — for zip-packaged functions — the per-function requirements.txt that python_instance_main.py:209-216 pip-installs, which is where an operator would pin it.

If you would rather keep older clients working anyway, the idiom is one method away: get_negative_ack_args() returns a dict to splat exactly so that an unconfigured argument is never passed at all, and a get_dead_letter_args() returning {} or {"dead_letter_policy": policy} would keep unconfigured functions on the old call shape. Given the images and CI already pin 3.13.0, documenting the requirement looks like the better trade.

[lhotari]

[QUALITY] test_policy_is_not_gated_on_subscription_type passes with the subscription-type gate fully restored

Removing the gate is right, and I agree with the reasoning in the thread. What I could not reproduce is the claim that this test leaves the intent "pinned rather than just untested".

I copied python_instance.py and put the gate from 5cc6207b061 back into get_dead_letter_policy(), deriving the consumer mode from function_details the way run() does at lines 143-152:

if mode not in (pulsar._pulsar.ConsumerType.Shared, pulsar._pulsar.ConsumerType.KeyShared):
  return None

All 33 tests in test_python_instance still pass. As a control, making get_dead_letter_policy() always return None fails 5 of them, so the suite is running and does have teeth — the gate is just invisible to it.

The reason is the fixture: _instance() sets only sink.topic and retryDetails, so source.subscriptionType, retainOrdering, retainKeyOrdering and processingGuarantees all keep their proto defaults. That resolves to SHARED = 0, which is precisely the case the removed gate allowed. The test's assertions are also identical to test_policy_built_from_retry_details, so it adds no signal beyond it.

What it does pin is the signature change: the old get_dead_letter_policy(consumer_type) would raise TypeError here. That is worth having, but it is a narrower guarantee than the name and comment claim.

To pin the behaviour, build the policy for an instance that actually selects a non-Shared subscription — retainOrdering=True, or processingGuarantees=EFFECTIVELY_ONCE, or source.subscriptionType=FAILOVER — and assert the policy is still produced. Since get_dead_letter_policy() no longer sees a consumer type at all, the durable place for that assertion is the subscribe() call in run() rather than the helper.

[lhotari]

[QUALITY] test_empty_dead_letter_topic_defers_to_client_default does not pin the or None, and the rationale in its comment is not what the client does

Two separate things here.

The test does not cover what it exists for. It asserts only assertIsNotNone(policy) and max_redeliver_count == 2. I changed dead_letter_topic=retry_details.deadLetterTopic or None to dead_letter_topic=retry_details.deadLetterTopic and all 33 tests in the module stayed green. Asserting policy.dead_letter_topic directly would at least pin the observable value.

The stated reason for the or None is not accurate. The comment (and the matching paragraph in the get_dead_letter_policy() docstring) says passing "" through "would override that with an invalid name". It would not: DeadLetterPolicyBuilder::deadLetterTopic("") is indistinguishable from never calling it — the impl field is a plain std::string defaulting to empty (DeadLetterPolicyBuilder.cc) — and ConsumerImpl.cc derives topic + "-" + subscriptionName + "-DLQ" whenever getDeadLetterTopic().empty(). So the or None is harmless and arguably clearer, but it is defensive rather than load-bearing.

Since the C++ policy exposes no presence flag, "unset" and "" are not distinguishable through the Python API at all, which is worth saying plainly instead of asserting a failure mode that cannot occur. Either drop the claim from the comment and docstring, or keep the or None and describe it as normalisation.

[lhotari]

[QUALITY] the new tests were inserted between @SuppressWarnings("deprecation") and the method it was suppressing for

On master this annotation sits directly above private FunctionConfig createFunctionConfig(), which calls the @Deprecated FunctionConfig.setAutoAck(true). The new tests were inserted between the two, so at head the annotation applies to testPythonFunctionAcceptsMessageRetriesAndDeadLetterTopic() (line 539) — which touches no deprecated API — and createFunctionConfig() (now line 596, setAutoAck(true) at line 618) is left unsuppressed.

The build compiles with -Xlint:deprecation (build-logic/conventions/src/main/kotlin/pulsar.java-conventions.gradle.kts:69), so this reintroduces the warning the annotation was added to silence. It is not an error, which is why CI stays green. Corroboration that this is the annotation's real target: every other @SuppressWarnings("deprecation") in the file — lines 80, 93, 139, 658 — sits on a method that calls setAutoAck.

Moving the new tests below createPythonFunctionConfig() (or just re-attaching the annotation) restores it.

[lhotari]

[QUALITY] duplicate import pulsar

The module already imports pulsar at line 41. The new TestDeadLetterPolicy tests reach the policy through instance.get_dead_letter_policy() and never reference pulsar. themselves, so this line can go.

[lhotari]

[QUALITY] no test exercises either modified subscribe() call

Every test in this class calls get_dead_letter_policy() directly. Nothing asserts that the policy actually reaches subscribe(), on either the topicsToSerDeClassName path (python_instance.py:186-192) or the inputSpecs/regex path (python_instance.py:211-218).

That is the coverage gap behind the other three test findings: the policy could be dropped from one of the two call sites, gated again on subscription type, or passed under a keyword the installed client does not accept, and this suite would stay green.

The file already has the pattern for it — the producer-config tests around lines 300-365 assert on kwargs captured from a mocked client. Applying the same shape to a mocked pulsar_client.subscribe would cover the two call sites, and would give the subscription-type intent somewhere durable to live now that get_dead_letter_policy() no longer sees a consumer type.