[Python Functions] Python instance runtime silently ignores consumerCryptoFailureAction

[Zhianii]

Search before reporting

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

Motivation

The Python function runtime silently ignores ConsumerSpec.cryptoSpec.consumerCryptoFailureAction: the Java runtime applies it, the Python runtime drops it, and the pinned Python client already supports the corresponding subscribe() parameter — so this is purely an instance-runtime gap.

A user hits it by configuring a crypto failure action on an encrypted input topic:

inputSpecs:
  "persistent://public/default/encrypted-input":
    cryptoConfig:
      cryptoKeyReaderClassName: "myapp.MyKeyReader"
      consumerCryptoFailureAction: CONSUME

• Expected (what the Java runtime does): undecryptable messages are delivered to the function in cleartext; with DISCARD they are dropped.
• Observed (Python runtime): the setting has no effect. The client falls back to its default FAIL — the message is logged as an error and redelivered indefinitely instead of being passed through or dropped.

The failure is silent: pulsar-admin accepts the config, functions get reports it back faithfully, and nothing at runtime indicates it was dropped. Unset config is unaffected (proto default FAIL = client default, so runtimes behave identically until the field is set).

Reproducing the issue

On master @ 8ae58a1:

1. The field reaches the instance: FunctionConfig.inputSpecs.<topic>.cryptoConfig.consumerCryptoFailureAction (CryptoConfig.java L45) is serialized by CryptoUtils.convert (CryptoUtils.java L57-62) into ConsumerSpec.cryptoSpec (Function.proto L148-149).
2. The Java runtime applies it: PulsarSource.java L87-88 — cb.cryptoFailureAction(conf.getConsumerCryptoFailureAction()).
3. The Python runtime drops it: python_instance.py L202-216 builds consumer_args with only crypto_key_reader; get_crypto_reader (L605-616) reads only the reader class/config. A grep for CryptoFailureAction in pulsar-functions/instance/src/main/python/ matches only the generated Function_pb2.py.

Additional information

• No client-library change needed: the pinned pulsar-client-python = "3.13.0" (gradle/libs.versions.toml L22) exposes crypto_failure_action on subscribe() (v3.13.0 pulsar/__init__.py L1240, applied at L1419).
• Producer side is out of scope: producerCryptoFailureAction is likewise unread, but v3.13.0 create_producer() has no such parameter, so honoring it needs client support first.
• Related issues: [Python Functions] Python instance runtime silently ignores deadLetterTopic / maxMessageRetries #26397 (retryDetails), [Python Functions] consumerProperties, poolMessages and messagePayloadProcessorSpec are ignored #26410, [Python Functions] negativeAckRedeliveryDelayMs is ignored, so the client default always applies #26411 cover other silently-dropped Python-runtime config; none covers crypto failure actions. The parity audits [Go Functions] Close the configuration parity gaps with the Java and Python runtimes (master issue) #26404 / [Python Functions] Close the configuration parity gaps with the Java runtime (master issue) #26412 mark Python cryptoSpec (consumer) as applied — true for the key reader ([improve][fn] Support e2e cryption in python instance  #18738), not for the failure action; this issue corrects that row.

Solution

In python_instance.py's input-consumer setup (L202-216), map consumer_conf.cryptoSpec.consumerCryptoFailureAction onto pulsar.ConsumerCryptoFailureAction — mirroring the existing CompressionType mapping at L389-398 — and pass it as crypto_failure_action in consumer_args. Unset config keeps the client default, so behavior is unchanged for functions that don't configure the field.

Test: extend pulsar-functions/instance/src/test/python/test_python_instance.py, which already mocks the client and asserts producer kwargs (_create_producer_kwargs); assert subscribe receives the mapped crypto_failure_action for each enum value.

• I'm willing to submit a PR!