[fix][fn] Honour deadLetterTopic and maxMessageRetries in the Python function runtime

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

Modifications

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.

Two cases cannot mirror Java exactly. Both warn rather than failing the instance or silently doing nothing:

1. maxMessageRetries == 0. PulsarSource accepts >= 0, but the Python client's ConsumerDeadLetterPolicy raises ValueError for a redelivery count below 1 (pulsar/__init__.py L761-762). Attaching no policy is the only available behaviour; the warning names the dead letter topic that will not receive messages. Raising here would take down a function that 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.
• Full file: 12/12 pass via pulsar-functions/instance/src/scripts/run_python_instance_tests.sh.
• Confirmed the 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.

Documentation

• doc-required

The Pulsar Functions documentation states that --dead-letter-topic and --max-message-retries are supported without qualifying which runtimes honour them. #6084 was closed in 2020 by a docs update saying Python does not support this; with this change that note needs revisiting. I have not made the apache/pulsar-site change — 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.