[Python Functions] Python instance runtime silently ignores deadLetterTopic / maxMessageRetries #26397

Search before reporting

• I searched in the issues and found nothing similar that is still open.

Prior art — two related issues exist, both closed, neither resolving this:

• [v2.5.0] Functions dead-letter-topic parameter not supported #6084 "[v2.5.0] Functions dead-letter-topic parameter not supported" — closed as completed in June 2020, but the resolution was a documentation update (Update the function CLI description #6980), not an implementation. From that thread: "Python functions doesn't support this feature at all." Still true on master.
• DeadLetterTopics for python client #9741 "DeadLetterTopics for python client" — closed as stale/not-planned in Dec 2022 on the grounds that the Python client lacked DLQ support. That blocker no longer exists; see Solution.

Motivation

FunctionConfig accepts maxMessageRetries and deadLetterTopic, and both are carried into the instance as FunctionDetails.retryDetails (Function.proto L58-61, L91). The Java runtime honours them. The Python runtime silently ignores them.

python_instance.py builds its consumer arguments with no DLQ policy (L201-209):

      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
      }

grep -i "dead_letter\|retryDetails" pulsar-functions/instance/src/main/python/python_instance.py returns 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 config back faithfully, and at runtime nothing is ever routed to the DLQ. Users discover it only when they go looking for messages that never arrived. Infrastructure-as-code makes this worse — e.g. the Terraform provider exposes dead_letter_topic and max_message_retries on pulsar_function with no runtime caveat, so the config applies cleanly and does nothing.

This matters more for Python than for Java, because the Python instance negative-acknowledges on any user exception (L282-286):

        except Exception as e:
          Log.exception("Exception while executing user method")
          self.stats.incr_total_user_exceptions(e)
          # If function throws exception then send neg ack for input message back to broker
          msg.consumer.negative_acknowledge(msg.message)

Since nack increments the redelivery count, DLQ routing would work correctly if a policy were attached. Without one, a message that can never succeed — a payload failing schema validation, say — is redelivered indefinitely at the client's default nack delay, with no exit path other than the function catching the error and hand-rolling a DLQ producer.

Solution

The blocker cited in #9741 is gone: pulsar-client-python now supports DLQ via ConsumerDeadLetterPolicy and Client.subscribe(dead_letter_policy=...) (pulsar/init.py — ConsumerDeadLetterPolicy at L738, dead_letter_policy parameter at L1239, applied at L1417-1418).

The config already reaches the instance in the protobuf, so this should be a small change local to setup_consumer():

      dead_letter_policy = None
      if self.instance_config.function_details.HasField("retryDetails"):
        retry = self.instance_config.function_details.retryDetails
        if retry.maxMessageRetries > 0:
          dead_letter_policy = ConsumerDeadLetterPolicy(
            max_redeliver_count=retry.maxMessageRetries,
            dead_letter_topic=retry.deadLetterTopic or None,
          )

      consumer_args = {
        ...
        "dead_letter_policy": dead_letter_policy,
      }

(HasField is already the idiom here — see L210 for receiverQueueSize.)

Points worth settling in review:

1. ConsumerDeadLetterPolicy raises ValueError unless max_redeliver_count >= 1, so maxMessageRetries <= 0 must mean "attach no policy" rather than being passed through.
2. deadLetterTopic is optional client-side and defaults to <topic>-<subscription>-DLQ. Whether that matches the Java runtime's behaviour when only maxMessageRetries is set should be confirmed so the two runtimes don't diverge.
3. There are three subscribe() call sites in setup_consumer() (L178, L214, L219); all need the policy.
4. DLQ requires a Shared or Key_Shared subscription. retain_ordering selects Failover, so that combination should warn rather than silently no-op — which is the same class of bug as this issue.

Alternatives

1. Leave it to user code — catch the exception in process() and publish to a DLQ topic via an explicitly created producer. This works and is what we're doing today, but every Python function author reimplements it, and it leaves --dead-letter-topic accepted-but-inert.
2. Reject the config for Python functions at submission time, turning silent no-op into a clear error. Strictly worse than implementing the feature, but far better than the status quo, and it could ship as an interim guard if the full implementation needs a PIP.

Anything else?

Verified against master at 00a6badafc62. No behaviour change for functions that don't set retryDetails, so this should be backportable.

Are you willing to submit a PR?

• I'm willing to submit a PR!