[fix][fn] Honour consumerCryptoFailureAction in the Python function runtime - #26490
Open
Zhianii wants to merge 1 commit into
Open
[fix][fn] Honour consumerCryptoFailureAction in the Python function runtime#26490Zhianii wants to merge 1 commit into
Zhianii wants to merge 1 commit into
Conversation
…untime Forward ConsumerSpec.cryptoSpec.consumerCryptoFailureAction to the Python client's subscribe() call, mirroring how the Java runtime applies it in PulsarSource. The action was previously accepted by pulsar-admin, reported back by `functions get` and then silently dropped, so the client default (FAIL) always applied and a configured DISCARD or CONSUME was ignored. The argument is omitted when no cryptoSpec is configured - subscribe() validates it with _check_type, which rejects None. When a cryptoSpec is present the action is always forwarded; the proto3 enum default (FAIL) equals the client default, so unset behavior is unchanged. The producer-only SEND value falls back to FAIL, like the compression mapping treats values without a consumer equivalent. Fixes apache#26481 Assisted-by: Claude Code
Contributor
There was a problem hiding this comment.
🟡 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 consumersubscribe()kwargs. - Map proto
CryptoSpec.FailureActionvalues topulsar.ConsumerCryptoFailureAction, falling back toFAILfor unsupportedSEND. - 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.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Fixes #26481
Motivation
ConsumerSpec.cryptoSpec.consumerCryptoFailureActiontells the client what to do when an input message cannot be decrypted (FAIL/DISCARD/CONSUME). The Java runtime applies it inPulsarSource(cb.cryptoFailureAction(...)), but the Python runtime built itsconsumer_argswith only the crypto key reader and never read the action: the config was accepted bypulsar-admin, reported back faithfully byfunctions get, and silently dropped, so the client default (FAIL) always applied and a configuredDISCARDorCONSUMEwas ignored.The pinned
pulsar-client-python(3.13.0) already exposescrypto_failure_actiononsubscribe(), 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: newget_crypto_failure_action_args(consumer_conf)builds thecrypto_failure_actionargument as a dict splatted intosubscribe(), following theget_negative_ack_argspattern:cryptoSpecconfigured → the argument is omitted (subscribe()validates it with_check_type, which rejectsNone), so every function without crypto keeps today's call shape;cryptoSpecpresent → the action is always forwarded; the proto3 enum default (FAIL) equals the client default, so an unset action leaves behavior unchanged whileDISCARDandCONSUMEnow reach the client;SENDvalue falls back toFAIL, like the existing compression mapping treats values without a consumer equivalent.Verifying this change
This change added tests and can be verified as follows:
TestConsumerCryptoFailureActioninpulsar-functions/instance/src/test/python/test_python_instance.py(same suite and mock-client style as the existingTestNegativeAckRedeliveryDelay/TestSinkProducerBatchingConfig): action omitted without a cryptoSpec, unset action keeps the client default,DISCARD/CONSUMEforwarded,SENDfalls back toFAIL, and the result is splattable into thesubscribe()kwargs.AttributeError: 'PythonInstance' object has no attribute 'get_crypto_failure_action_args'(verified by stashing the instance change) and pass with it.pulsar-functions/instance/src/scripts/run_python_instance_tests.sh: 36 tests pass (33 intest_python_instanceincluding the 6 new ones).Run locally with
pulsar-client 3.8.0(samecrypto_failure_actionAPI 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