apache / apache/pulsar

[Python Functions] Python instance runtime silently ignores consumerCryptoFailureAction

Open
#26,481 0 comments 0 reactions 0 assignees View on GitHub
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

Open the contributing 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

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.