[fix][fn] Honour negativeAckRedeliveryDelayMs in the Python function runtime

[david-streamlio]

Fixes #26411
Master Issue: #26412

Motivation

SourceSpec.negativeAckRedeliveryDelayMs sets how long the broker waits before redelivering a negatively acknowledged message. The Python runtime negatively acknowledges on failure:

# python_instance.py:286, 299, 310
msg.consumer.negative_acknowledge(msg.message)

but never configured the delay, so subscribe()'s default of 60000 applied regardless of what the function was created with. A function configured for fast retry, or for a long back-off from a struggling downstream, silently got one minute either way.

The Java runtime applies it. The Go runtime has the same gap, tracked at #26409.

Modifications

Add PythonInstance.get_negative_ack_args() and splat its result into all three subscribe() call sites — the topicsToSerDeClassName loop and both branches of the inputSpecs loop.

Two details worth a reviewer's attention, both of which shaped the API rather than being incidental:

1. Unset reads as zero. The field is a proto3 scalar with no presence, so a function that never set it arrives with 0. Only a positive value is forwarded, leaving the client default in place otherwise — the same guard JavaInstanceRunnable applies (if (sourceSpec.getNegativeAckRedeliveryDelayMs() > 0)). Forwarding 0 would mean immediate redelivery rather than the 60s default, which is not what an unconfigured function should get.

2. The argument is omitted, not passed as None. subscribe() validates this one with _check_type(int, negative_ack_redelivery_delay_ms, ...) rather than _check_type_or_none — unlike the neighbouring unacked_messages_timeout_ms, which the existing code does pass None to. Passing None here would raise for every function that does not configure the field. Returning a dict to splat keeps that decision in one place, since the first call site passes explicit keywords while the other two build a consumer_args dict.

The helper is a method rather than inline code so it is directly testable, matching get_dead_letter_policy() added in #26400.

Verifying this change

• Make sure that the change passes the CI checks.

This change added tests and can be verified as follows:

• Four unit tests in test_python_instance.py (TestNegativeAckRedeliveryDelay): a positive delay is forwarded; an unset field is omitted; an explicit 0 is omitted; and the return value is a dict carrying exactly the keyword subscribe() expects, since it is consumed via ** and .update().
• Full file: 8/8 pass via pulsar-functions/instance/src/scripts/run_python_instance_tests.sh.
• Confirmed the tests are not vacuous: making get_negative_ack_args() return {} unconditionally fails 2 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 negativeAckRedeliveryDelayMs is unaffected: the argument is omitted and the client default applies exactly as before.

Documentation

• doc-required
• doc-not-needed
• doc
• doc-complete

The configuration option is already documented; this makes the Python runtime honour it.