[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.

[david-streamlio]

Closing the loop for the record: @lhotari independently re-checked this one rather than taking the reply on trust, and confirmed both halves — 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.

So localrun now fails fast the same way Java does. Thanks for catching it.

[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]

@lhotari independently confirmed the removal is the right call, for the reason you gave: the Java runtime forwards a configured policy unconditionally, so gating in the runtime created a second support matrix that would drift from the client.

One correction to my reply above, since it was overstated at the time. I said the replacement test left the intent "pinned rather than just untested" — it did not. @lhotari re-applied the old gate to a copy of get_dead_letter_policy() and the whole suite stayed green, because every FunctionDetails the tests build leaves retainOrdering, retainKeyOrdering, processingGuarantees and source.subscriptionType at their proto defaults, so the derived consumer type was always Shared — the one value the old gate allowed.

That is now genuinely fixed, in c3b683f: the assertion lives in TestConsumerSubscribeArgs, where the consumer type is actually derived, and drives it to both Failover and KeyShared. Mutation-tested in both directions — a gate sparing Shared/KeyShared and a gate blocking KeyShared only — and in each case it is the only failing test in the suite.

[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.

[david-streamlio]

@lhotari all six points are addressed in 94b4fcd — thanks, the review found real gaps rather than
style nits, and two of them were things I had asserted without checking.

• Client floor — documented in python/README.md, per your call. Verified against the client
  sources: v3.2.0 has neither dead_letter_policy nor ConsumerDeadLetterPolicy, v3.3.0 adds
  both. I also diffed every keyword the runtime passes against the 3.2.0 signature —
  dead_letter_policy is the only one missing, so 3.3.0 is exactly the floor this introduces.
• subscribe() coverage — extracted setup_consumers() to get the same seam setup_producer()
  has, and covered both call sites plus the regex variant against a mocked client.
• or None — removed. It could not change behaviour: deadLetterTopic("") and never calling it
  produce the same object. The old comment's rationale was wrong; the test now pins the runtime's
  actual half of the contract.
• Subscription-type test — you were right that the gate restores green. It used a default
  instance, which selects Shared, the one type the gate allowed. Now runs the four FunctionDetails
  inputs that select a consumer type.
• @SuppressWarnings and the duplicate import — both fixed.

I mutation-tested each rewritten test rather than just checking it passes; the specific mutations and
what fails under them are in the threads.

Verification: run_python_instance_tests.sh against a venv with the CI-pinned deps
(pulsar-client[all]==3.13.0, grpcio==1.78.0, protobuf==6.33.6) — 40 tests, all pass;
FunctionConfigUtilsTest — 48 tests, 0 failures; ./gradlew quickCheck clean.

One question for you in the client-floor thread: the README line is the first declared minimum
pulsar-client-python version in the repo, so if the project wants to state a higher supported
minimum than the 3.3.0 technical floor, that number should be yours.

The matching note for apache/pulsar-site#1214 is next — that PR has no reviewers assigned, so if you
have a moment for it there too it would unblock the docs half.

@freeznet when you have a chance — your two points were fixed back in 47df72c and lhotari confirmed
both independently, so this is only waiting on the standing CHANGES_REQUESTED being cleared.

[lhotari]

All six points from the previous round are addressed in 94b4fcd, and I checked each one rather than taking the replies on trust.

What I verified independently:

• The 3.3.0 floor is right. pulsar/__init__.py at tag v3.2.0 contains neither dead_letter_policy nor ConsumerDeadLetterPolicy; v3.3.0 adds both and guards with if dead_letter_policy:. I also diffed the subscribe() and create_producer() signatures between v3.2.0 and v3.13.0 against every keyword this runtime actually passes — dead_letter_policy is the only one 3.2.0 lacks, so the README's "this is the only thing setting the floor" holds.
• The or None removal is safe. ConsumerDeadLetterPolicy.__init__ forwards the topic only when it is not None, DeadLetterPolicyBuilder::deadLetterTopic is a plain assignment, and build() validates only maxRedeliverCount — so "" and unset really are indistinguishable, and the rewritten docstring now says so.
• The setup_consumers() extraction is behaviour-preserving. Diffing the moved block against the previous head gives zero non-blank differences, and none of the locals it takes with it (mode, position, properties, subscription_name, nack_args) are read later in run().
• The new tests have teeth. I ran test_python_instance against the CI-pinned client (37 tests, green) and reproduced the mutations described in the threads. Restoring the old gate inside get_dead_letter_policy() fails exactly 3 of the 4 subTests, with retainKeyOrdering correctly still passing. Dropping the policy from the topicsToSerDeClassName call site fails only that path's test; dropping it from the inputSpecs call site fails only the inputSpecs and regex tests. The two call sites are independently covered.

That leaves one residual and two small notes, all inline and none of them blocking: a subscription-type gate reintroduced in setup_consumers() — which is where it would go now — is still invisible to the suite, and closing that is one parameter on a helper you already have.

Thanks for the mutation testing in the replies; it made this round much faster to check.

[lhotari]

Thanks — this was a genuinely easy round to check, precisely because you named the mutation for each claim instead of just saying the tests pass. I reproduced them rather than taking them on trust, and they hold: the restored gate fails 3 of the 4 subTests with retainKeyOrdering correctly passing, and each call-site drop fails only its own path's test. I also re-derived the 3.3.0 floor and your "only keyword 3.2.0 lacks" result from the client sources, and diffed the setup_consumers() extraction against the previous head to confirm it is a pure move.

On the minimum version: keep 3.3.0. It is the technical floor, it is checkable against the client sources, and that is the kind of statement a bug fix should be making. A declared supported minimum is a project commitment rather than a fact about this change; if we want one it belongs on dev@, not in a runtime README. I have said the same in the thread.

Three inline comments this round, all small: one residual test gap where a subscription-type gate could come back in setup_consumers() unnoticed, one correction to the README's claim about CI reading the version catalog, and a missing blank line. None of them need to hold the PR.

[lhotari]

All six open points are closed in c3b683f5, and I checked each at that commit rather than reading the replies.

The one that mattered was the invisible subscription-type gate. test_policy_reaches_subscribe_for_a_non_shared_consumer_type now lives in TestConsumerSubscribeArgs, where setup_consumers() actually derives the consumer type, and it drives that type over all four selectors that decide it. I re-ran the module against pulsar-client[all]==3.13.0, grpcio==1.78.0, protobuf==6.33.6: 38 tests green at head; a gate sparing Shared/KeyShared fails 3 of the 4 subTests; a gate blocking KeyShared only fails the fourth. In both runs it is the only failing test in the module. Covering both non-default modes rather than the single case I suggested is the better call — a one-case version would have sailed past the second gate shape.

The other two were a README sentence and a blank line, both correct at head. The README now matches the mechanism it describes: the Docker images derive their version from the catalog, CI pins the same number separately in run_python_instance_tests.sh:34, and the two agree by maintenance. I re-checked each half.

Nothing open from me. The review decision on the PR is still in the "changes requested" state from the 26 August review, which an approval from me would not clear on its own — the author has since replied on both of those threads. Perhaps the "changes requested" review can be dismissed?