Add DLQ rule action to Schema Registry serdes#2310
Open
Justin Wang (Claimundefine) wants to merge 7 commits into
Open
Add DLQ rule action to Schema Registry serdes#2310Justin Wang (Claimundefine) wants to merge 7 commits into
Justin Wang (Claimundefine) wants to merge 7 commits into
Conversation
Justin Wang (Claimundefine)
requested review from
a team and
Matthew Seal (MSeal)
as code owners
July 20, 2026 05:07
|
🎉 All Contributor License Agreements have been signed. Ready to merge. |
There was a problem hiding this comment.
Pull request overview
This PR adds a Schema Registry “DLQ” rule action to the Python serdes, mirroring the Java client’s behavior: when a rule fails with onFailure=DLQ, the original record is teed to a configured dead-letter topic while still raising SerializationError. It also adds field redaction for tagged fields (to avoid plaintext leakage) and replay-skipping via the __rule.name header.
Changes:
- Introduces
DlqAction+FieldRedactionExecutor, including byte conversion, header population, redaction, and producer lifecycle behavior. - Captures original key/value via a
contextvars.ContextVarand threads that through rule execution so DLQ sends can use the “entry” payload/wire bytes. - Adds comprehensive unit + integration tests for sync/async Avro/JSON/Protobuf paths, replay behavior, redaction, and close/ownership semantics.
Reviewed changes
Copilot reviewed 16 out of 16 changed files in this pull request and generated no comments.
Show a summary per file
| File | Description |
|---|---|
| tests/schema_registry/test_dlq_action.py | Unit tests for DlqAction conversion, topic resolution, headers, redaction, replay detection |
| tests/schema_registry/_sync/test_dlq_serdes.py | Sync serde tests covering DLQ behavior across Avro/JSON/Protobuf, redaction, replay skip, close semantics |
| tests/schema_registry/_async/test_dlq_serdes.py | Async serde tests mirroring sync coverage for DLQ behavior, redaction, replay skip, close semantics |
| tests/integration/schema_registry/_sync/test_dlq.py | Sync integration test validating DLQ tee + replay-skip with a real Kafka cluster |
| tests/integration/schema_registry/_async/test_dlq.py | Async integration test validating DLQ tee + replay-skip with a real Kafka cluster |
| src/confluent_kafka/schema_registry/rules/dlq/dlq_action.py | Implements DlqAction + FieldRedactionExecutor (produce-to-DLQ, redaction, headering, config) |
| src/confluent_kafka/schema_registry/rules/dlq/init.py | Adds dlq rules package marker |
| src/confluent_kafka/schema_registry/common/serde.py | Adds DLQ header constants + contextvar-based original key tracking; extends RuleContext with original key/value |
| src/confluent_kafka/schema_registry/_sync/serde.py | Threads original-message support through rule execution; adds DLQ replay-skip; adds close() ownership semantics for registries |
| src/confluent_kafka/schema_registry/_sync/protobuf.py | Configures actions; captures/clears original key; passes wire bytes into READ encoding-phase rule execution |
| src/confluent_kafka/schema_registry/_sync/json_schema.py | Configures actions; captures/clears original key; passes wire bytes into READ encoding-phase rule execution |
| src/confluent_kafka/schema_registry/_sync/avro.py | Configures actions; captures/clears original key; passes wire bytes into READ encoding-phase rule execution |
| src/confluent_kafka/schema_registry/_async/serde.py | Async analogue of sync serde updates: original-message threading, DLQ replay-skip, aclose() ownership semantics |
| src/confluent_kafka/schema_registry/_async/protobuf.py | Async analogue: action configure, original key capture/clear, pass wire bytes for READ encoding-phase |
| src/confluent_kafka/schema_registry/_async/json_schema.py | Async analogue: action configure, original key capture/clear, pass wire bytes for READ encoding-phase |
| src/confluent_kafka/schema_registry/_async/avro.py | Async analogue: action configure, original key capture/clear, pass wire bytes for READ encoding-phase |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
Port the Java client's DlqAction to confluent-kafka-python. When a data-contract rule (CSFLE encryption, CEL condition, etc.) fails during serialize/deserialize and the rule's onFailure is set to "DLQ", the record is teed to a dead-letter-queue topic and a SerializationError is still raised (tee, not swallow). - DlqAction + FieldRedactionExecutor under rules/dlq/ - Capture original key/value for the DLQ record via a contextvar (thread- and task-safe), mirroring the Java client's ThreadLocal - Redact fields tagged by encryption rules before JSON-encoding so plaintext does not leak to the DLQ - Skip a rule on replay when the record carries the matching __rule.name header - Configure registered rule actions in each serde; close owns non-global registries Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
A single DlqAction is shared across every serde bound to the same RuleRegistry, so each serde constructor may call configure() concurrently. configure() mutated the shared _conf dict without holding the lock that _get_producer() uses while iterating it, so a serde constructed concurrently with the first DLQ send could trigger "RuntimeError: dictionary changed size during iteration" or build the producer from a half-written config. - configure() now mutates _conf and derives fields under self._lock; _parse_conf() stays lock-free (called with the lock held, or from __init__ before sharing) to avoid re-entrant deadlock. - close() detaches the producer under the lock, then flushes outside it (flush can block indefinitely with message.timeout.ms=0 and must not stall other callers). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
The Schema Registry DLQ rule action captures the value's original key via a contextvar that the key serde stashes for the value serde to read. This assumes the key is processed before the value, which holds on produce (SerializingProducer) and in the Java client, but DeserializingConsumer and DeserializingShareConsumer deserialized the value first. As a result, a READ-rule failure teed a DLQ record with the *previous* message's key (or None for the first message) instead of the current one. Deserialize the key before the value in both consumers so the value serde sees the current message's key. This aligns the consume path with the produce path and the Java client. As a side effect, when both key and value fail to deserialize, the key error is now surfaced instead of the value error. Also extract the classic consumer's deserialization into a _deserialize() method (mirroring DeserializingShareConsumer) so it can be unit-tested without a broker, and correct the DLQ docstrings that claimed the value serde clears the stashed key "on entry" (it clears on exit, after use). - Reorder DeserializingConsumer / DeserializingShareConsumer to key-first - Add tests/test_DeserializingConsumer.py; update share-consumer tests that pinned the old value-first order - Update dlq_action / common serde docs and CHANGELOG Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Follow-up review changes to the Schema Registry DLQ action: - Add a durability contract to the DlqAction docstring: with the default (global) RuleRegistry the DLQ is best-effort (records teed just before process exit may be dropped). For durability, set dlq.auto.flush=true or give the serde its own RuleRegistry and call close()/aclose() on shutdown. - Reword DLQ comments and docstrings to describe the Python implementation directly instead of referencing the Java client, across dlq_action.py, schema_registry/_async|_sync/serde.py, common/serde.py, deserializing_consumer.py, and the DLQ tests. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Three fixes to the DLQ rule action's shutdown and redaction behavior, all exercised by the DLQ_DESIGN "dedicated registry" durable-shutdown path: 1. FieldEncryptionExecutor.close() referenced a non-existent self.client and raised AttributeError. On the recommended shutdown path (a dedicated registry holding a FieldEncryptionExecutor + DlqAction), the serde's close loop hit this first, so the DlqAction never flushed -- buffered DLQ records were lost and the KMS client leaked. Delegate to the wrapped EncryptionExecutor, which owns the client. 2. BaseSerde.close()/aclose() closed every executor then every action with no error isolation, so one failing close skipped the rest -- including the DlqAction flush. Flush actions first and wrap each close in try/except, so a buggy executor can't cost buffered DLQ records. 3. Redaction mutated the caller's message in place, so a WRITE failure (e.g. a transient KMS/DEK outage) silently replaced the app's live PII fields with "<REDACTED>"; retrying the same object would then serialize masked data. Redact a deep copy instead. The DLQ record bytes are identical, so this is a local-only divergence from Java with no cross-client effect. Also document the ENCRYPT_PAYLOAD-on-WRITE plaintext exposure prominently (a payload-level rule has no field tags to redact, so a WRITE failure tees the full plaintext record; matches Java, so behavior is unchanged). Adds regression tests: FieldEncryptionExecutor.close() does not raise; the real-executor durable-shutdown path flushes the DLQ; a failing executor is isolated and the DLQ still flushes; and redaction leaves the caller's object intact. (_sync regenerated from _async via unasync.) Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
The synthetic redact rule built in _redact_fields is labelled with
DlqAction.TYPE ("DLQ"), matching the Java client. This is harmless: the
rule drives FieldRedactionExecutor.transform() directly (never looked up
by type), and in the single-rule context are_transforms_with_same_tag()
is unreachable, so rule.type is never read. Add a comment so the value
is not mistaken for a bug and stays in sync with Java for any future
type-based dispatch.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Condense the over-verbose DLQ docstrings and inline comments to fit the surrounding code (e.g. DlqAction's ~110-line class docstring down to ~20, multi-line inline comments to one or two lines each), with no behavior change. Also move the DLQ note in the changelog from Fixes to Enhancements (it is a new feature) and slim the key-before-value Fixes entry. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
airlock-confluentinc
Bot
force-pushed
the
dlq-action
branch
from
July 23, 2026 02:42
2c90990 to
4a7bcde
Compare
|
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.
Port the Java client's DlqAction to confluent-kafka-python. When a data-contract rule (CSFLE encryption, CEL condition, etc.) fails during serialize/deserialize and the rule's onFailure is set to "DLQ", the record is teed to a dead-letter-queue topic and a SerializationError is still raised (tee, not swallow).
What
Checklist
References
JIRA:
DGS-14069
Test & Review
Open questions / Follow-ups