[Python Functions] Python instance runtime silently ignores consumerCryptoFailureAction
- Dominant language
- Java
- Stars
- 15.3k
- Forks
- 3.8k
- Avg merge
- 1d 14h
- Merged PRs (30d)
- 160
Description
### Search before reporting
- [X] I searched in the [issues](https://github.com/apache/pulsar/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:
```yaml
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..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**: #26397 (retryDetails), #26410, #26411 cover other silently-dropped Python-runtime config; none covers crypto failure actions. The parity audits #26404 / #26412 mark Python `cryptoSpec` (consumer) as applied — true for the key reader (#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.
- [X] I'm willing to submit a PR!
Contributor guide
Research direction
Start in pulsar-functions/instance/src/main/python/python_instance.py at the input-consumer setup around L202-216, then compare the existing CompressionType mapping around L389-398. Extend pulsar-functions/instance/src/test/python/test_python_instance.py, which already mocks the client, and verify subscribe receives the expected crypto_failure_action for each enum value while unset configuration retains the default.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- python
- Domain
- backend, distributed-systems
- Issue type
- Bug
- Difficulty
- 3/5
- Estimated time
- 1-2 days
- Activity status
- Active
- Clarity
- Clearly specified
- Newbie friendliness
- 84/100