[Zhianii]

Fixes #26481

Motivation

ConsumerSpec.cryptoSpec.consumerCryptoFailureAction tells the client what to do when an input message cannot be decrypted (FAIL / DISCARD / CONSUME). The Java runtime applies it in PulsarSource (cb.cryptoFailureAction(...)), but the Python runtime built its consumer_args with only the crypto key reader and never read the action: the config was accepted by pulsar-admin, reported back faithfully by functions get, and silently dropped, so the client default (FAIL) always applied and a configured DISCARD or CONSUME was ignored.

The pinned pulsar-client-python (3.13.0) already exposes crypto_failure_action on subscribe(), so no client change is needed — this closes the runtime gap only. The producer side (producerCryptoFailureAction) is still unread but has no client parameter to map onto and stays out of scope.

Modifications

• pulsar-functions/instance/src/main/python/python_instance.py: new get_crypto_failure_action_args(consumer_conf) builds the crypto_failure_action argument as a dict splatted into subscribe(), following the get_negative_ack_args pattern:
  • no cryptoSpec configured → the argument is omitted (subscribe() validates it with _check_type, which rejects None), so every function without crypto keeps today's call shape;
  • cryptoSpec present → the action is always forwarded; the proto3 enum default (FAIL) equals the client default, so an unset action leaves behavior unchanged while DISCARD and CONSUME now reach the client;
  • the producer-only SEND value falls back to FAIL, like the existing compression mapping treats values without a consumer equivalent.

Verifying this change

This change added tests and can be verified as follows:

• New TestConsumerCryptoFailureAction in pulsar-functions/instance/src/test/python/test_python_instance.py (same suite and mock-client style as the existing TestNegativeAckRedeliveryDelay / TestSinkProducerBatchingConfig): action omitted without a cryptoSpec, unset action keeps the client default, DISCARD / CONSUME forwarded, SEND falls back to FAIL, and the result is splattable into the subscribe() kwargs.
• The new tests fail on unpatched master with AttributeError: 'PythonInstance' object has no attribute 'get_crypto_failure_action_args' (verified by stashing the instance change) and pass with it.
• Full suite via pulsar-functions/instance/src/scripts/run_python_instance_tests.sh: 36 tests pass (33 in test_python_instance including the 6 new ones).

Run locally with pulsar-client 3.8.0 (same crypto_failure_action API surface as the pinned 3.13.0; CI runs the pinned version).

Does this pull request potentially affect one of the following parts:

If the box was checked, please highlight the changes

• 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
• The admin CLI options
• The metrics
• Anything that affects deployment

[Copilot]

🟡 Changes recommended

The consumer setup still calls get_crypto_reader(consumer_conf.cryptoSpec) unconditionally, which can trigger noisy startup errors for non-crypto consumers due to parsing an empty default cryptoKeyReaderConfig.

Once you've addressed the issues Copilot identified, you can request another Copilot review.

Pull request overview

This PR fixes a Python Functions runtime parity gap by honoring ConsumerSpec.cryptoSpec.consumerCryptoFailureAction when subscribing to input topics, aligning Python behavior with the Java runtime and ensuring configured DISCARD/CONSUME actions reach the Pulsar Python client.

Changes:

• Add get_crypto_failure_action_args(consumer_conf) and wire it into input consumer subscribe() kwargs.
• Map proto CryptoSpec.FailureAction values to pulsar.ConsumerCryptoFailureAction, falling back to FAIL for unsupported SEND.
• Add Python instance tests covering omitted/forwarded/fallback behavior and ensuring kwargs are splattable into subscribe().

File summaries

File	Description
pulsar-functions/instance/src/main/python/python_instance.py	Forwards consumer crypto failure action into subscribe() kwargs via a new helper.
pulsar-functions/instance/src/test/python/test_python_instance.py	Adds unit tests validating crypto failure action forwarding and omission behavior.

Review details

• Files reviewed: 2/2 changed files
• Comments generated: 1
• Review effort level: Lite

---

💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.

[merlimat]

Thanks @Zhianii. I pushed two commits to bring this up to date:

• Merged master. [fix][fn] Support deadLetterTopic and maxMessageRetries for Python functions #26400 had moved the subscribe calls from run() into setup_consumers(), which caused the conflict, so the consumer_args.update(self.get_crypto_failure_action_args(consumer_conf)) line now lives there.
• Added two tests to TestConsumerCryptoFailureAction that go through setup_consumers() using the TestConsumerSubscribeArgs._subscribe_kwargs(configure=...) harness. One checks that a configured DISCARD reaches subscribe() on both the plain and regex inputSpecs paths. The other checks that nothing is passed when there's no cryptoSpec. The helper-level tests alone wouldn't catch the argument being dropped at the call site.
• Added a docstring note that the Java runtime rejects SEND for a consumer, while the Python runtime falls back to FAIL.

All Python instance tests pass locally with pulsar-client 3.13.0.