DGS-24222 Add support for inline validation rules by rayokota · Pull Request #2326 · confluentinc/confluent-kafka-python

Robert Yokota (rayokota) · 2026-08-11T23:11:05Z

What

Ports inline validation rules from the JVM client
(schema-registry#4291) to Python.
Schemas can now carry CHECK-style constraints that are evaluated at serialize time.

For Avro/JSON Schema, rules live in the confluent:rules property on records/objects and
their fields; for Protobuf, in the existing Meta extension on messages and fields.
New validate_message() in common/{avro,json_schema,protobuf}.py, each mirroring the
transform() walker already in that module. Every rule encountered is evaluated and all
violations are collected (no fail-fast by default), each with a dotted path
(addr.zip, tags[3], scores["foo"]).
New CelValidator (rules/cel/cel_validator.py) evaluates a rule with this and now
bound; a rule must return a bool (False = failed) or a str (non-empty = failed, with that
string as the message).
Failures are aggregated into a single SerializationError listing every violation, with
text identical to the JVM client's.
New serializer configs:
- validation.rules.execution — DISABLED (default), BEFORE_DOMAIN_RULES, AFTER_DOMAIN_RULES
- validation.rules.fail.fast — stop at the first violation (default False)
- validation.rules.executor — override the executor instance (defaults to CelValidator)

Checklist

[Y] Contains customer facing changes? Including API/behavior changes
[Y] Did you add sufficient unit test and/or integration test coverage for this PR?
- If not, please explain why it is not required

References

JIRA:

Test & Review

Open questions / Follow-ups

confluent-cla-assistant · 2026-08-11T23:11:16Z

🎉 All Contributor License Agreements have been signed. Ready to merge.
_{Please push an empty commit if you would like to re-run the checks to verify CLA status for all contributors.}

Copilot

Pull request overview

Adds inline schema validation rules (CHECK-style constraints) across Avro, JSON Schema, and Protobuf serialization in confluent-kafka-python, including a CEL-backed rule executor and serializer wiring to run validation before/after domain rules with optional fail-fast behavior.

Changes:

Introduces validate_message() walkers for Avro/JSON Schema/Protobuf to traverse messages and collect validation violations with stable dotted paths.
Adds a CEL-based CelValidator executor with per-expression program caching and this/now bindings.
Wires new serializer configs (validation.rules.*) into sync/async Avro, JSON, and Protobuf serializers and adds tests covering walkers, executor behavior, and serde integration.

Reviewed changes

Copilot reviewed 19 out of 21 changed files in this pull request and generated 2 comments.

Show a summary per file

File	Description
tests/schema_registry/test_validate_message.py	Unit tests for walker dispatch/pathing across Avro/JSON/Protobuf.
tests/schema_registry/test_cel_validator.py	Unit tests for CEL executor semantics, error surfaces, protobuf conversion, and caching.
tests/schema_registry/data/proto/validation_widget.proto	Protobuf fixtures defining inline rules via Meta extensions.
tests/schema_registry/data/proto/validation_widget_pb2.py	Generated pb2 fixture matching the proto validation test schema.
tests/schema_registry/_sync/test_validation_serdes.py	Sync serializer-level validation wiring tests and config validation.
tests/schema_registry/_async/test_validation_serdes.py	Async serializer-level validation wiring tests and config validation.
src/confluent_kafka/schema_registry/rules/cel/cel_validator.py	CEL-backed `ValidationRuleExecutor` implementation with program caching.
src/confluent_kafka/schema_registry/confluent/meta.proto	Extends Meta with repeated `rules` and adds `Rule` message.
src/confluent_kafka/schema_registry/confluent/meta_pb2.py	Regenerated descriptor bytes for updated Meta/Rule schema.
src/confluent_kafka/schema_registry/common/serde.py	Adds validation rule model/types, parsing, default executor resolution, and evaluation helper.
src/confluent_kafka/schema_registry/common/protobuf.py	Adds Protobuf validation walker reading Meta extensions and producing violations.
src/confluent_kafka/schema_registry/common/json_schema.py	Adds JSON Schema validation walker honoring combined keywords and `$ref` resolution.
src/confluent_kafka/schema_registry/common/avro.py	Adds Avro validation walker honoring unions/arrays/maps/records with skip-on-null.
src/confluent_kafka/schema_registry/_sync/serde.py	Adds sync base serde validation config parsing and aggregated error raising.
src/confluent_kafka/schema_registry/_sync/protobuf.py	Wires validation execution phases into sync Protobuf serialization.
src/confluent_kafka/schema_registry/_sync/json_schema.py	Wires validation execution phases into sync JSON serialization.
src/confluent_kafka/schema_registry/_sync/avro.py	Wires validation execution phases into sync Avro serialization.
src/confluent_kafka/schema_registry/_async/serde.py	Adds async base serde validation config parsing and aggregated error raising.
src/confluent_kafka/schema_registry/_async/protobuf.py	Wires validation execution phases into async Protobuf serialization.
src/confluent_kafka/schema_registry/_async/json_schema.py	Wires validation execution phases into async JSON serialization.
src/confluent_kafka/schema_registry/_async/avro.py	Wires validation execution phases into async Avro serialization.

Files not reviewed (2)

src/confluent_kafka/schema_registry/confluent/meta_pb2.py: Generated file
tests/schema_registry/data/proto/validation_widget_pb2.py: Generated file

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

Copilot

Pull request overview

Copilot reviewed 24 out of 27 changed files in this pull request and generated 1 comment.

Files not reviewed (3)

src/confluent_kafka/schema_registry/confluent/meta_pb2.py: Generated file
tests/schema_registry/data/proto/map_widget_pb2.py: Generated file
tests/schema_registry/data/proto/validation_widget_pb2.py: Generated file

Suppressed comments (4)

src/confluent_kafka/schema_registry/_sync/avro.py:265

These three new customer-facing serializer settings are absent from this class's Configuration properties table, and the JSON Schema, Protobuf, and async serializer tables omit them as well. Document the execution values, defaults, fail-fast behavior, and executor contract so users can configure the feature through the public API documentation.

        'validation.rules.execution': ValidationRulesExecution.DISABLED,
        'validation.rules.fail.fast': False,
        'validation.rules.executor': None,

src/confluent_kafka/schema_registry/rules/cel/cel_validator.py:23

Avro/JSON boolean values are converted through _value_to_cel, which checks int before bool; because bool subclasses int, True/False (including values nested in records) become CEL IntType values. Boolean constraints such as this == true or this.active will therefore fail or raise instead of evaluating as booleans. Preserve booleans before the integer branch in the shared recursive converter.

from confluent_kafka.schema_registry.rules.cel.cel_executor import _value_to_cel

src/confluent_kafka/schema_registry/common/avro.py:233

expand_schema() cannot inline recursive named types indefinitely, so self-references remain string schema names. This walker only dispatches lists and dicts; when it reaches a recursive child such as next: ["null", "Node"], the selected "Node" branch falls through as a primitive and rules on all deeper nodes are skipped. Resolve named string references through a schema-definition map before dispatching.

    if isinstance(schema, dict):
        schema_type = schema.get("type")

src/confluent_kafka/schema_registry/_sync/avro.py:481

The serializer tests exercise both modes only when there are no domain rules, so they cannot detect whether validation actually runs on the pre- or post-transformation value. Add a domain transform that changes a value across a validation boundary and assert that BEFORE_DOMAIN_RULES and AFTER_DOMAIN_RULES produce opposite outcomes; cover the mirrored async path as well.

            if self._validation_enabled(ValidationRulesExecution.BEFORE_DOMAIN_RULES):
                self._validate_inline_rules(expanded_parsed_schema, value)

Copilot

Pull request overview

Copilot reviewed 26 out of 29 changed files in this pull request and generated 2 comments.

Files not reviewed (3)

src/confluent_kafka/schema_registry/confluent/meta_pb2.py: Generated file
tests/schema_registry/data/proto/map_widget_pb2.py: Generated file
tests/schema_registry/data/proto/validation_widget_pb2.py: Generated file

Suppressed comments (4)

src/confluent_kafka/schema_registry/common/json_schema.py:312

For an external $ref selected by oneOf/anyOf, _validate_subschema returns only the referenced contents and discards the resolver returned by lookup. This recursion therefore uses the root resolver, so a relative $ref nested inside the external resource resolves against the wrong base URI (or raises NoSuchResource). Preserve and pass the resolved resource's resolver along with the selected schema.

                resolved = _validate_subschema(subschema, message, ref_registry, ref_resolver)
                if resolved is not None:
                    _validate_message(executor, resolved, ref_registry, ref_resolver, path, message, fail_fast, out)

src/confluent_kafka/schema_registry/rules/cel/cel_validator.py:100

This cache is unbounded, so a long-lived serializer retains every distinct validation expression it encounters across schema versions. The upstream validator caps its cache at 1,000 entries; use a bounded LRU cache here as well to prevent schema churn from causing permanent memory growth.

        self.programs = {}

tests/schema_registry/_sync/test_validation_serdes.py:145

This test gives both modes the same no-domain-rule path, so it cannot verify the defining before-vs-after behavior or catch swapped/misplaced hooks. Add a domain transformation that changes a failing value into a passing one and assert that BEFORE_DOMAIN_RULES fails while AFTER_DOMAIN_RULES succeeds; cover each serializer because the phase hooks are implemented separately.

@pytest.mark.parametrize("mode", ["BEFORE_DOMAIN_RULES", "AFTER_DOMAIN_RULES"])
def test_avro_both_modes_validate_when_no_domain_rules_exist(mode):
    ser = avro_serializer(**{'validation.rules.execution': mode})
    with pytest.raises(SerializationError, match="agePositive"):
        ser({"age": -5, "name": "Alice"}, ser_ctx())

tests/schema_registry/_async/test_validation_serdes.py:147

As written, both modes execute through the same no-domain-rule path, so this does not test their ordering and would still pass if the async phase hooks were reversed. Add a domain transformation that turns a failing value into a passing one, then assert BEFORE_DOMAIN_RULES fails and AFTER_DOMAIN_RULES succeeds for each async serializer.

@pytest.mark.parametrize("mode", ["BEFORE_DOMAIN_RULES", "AFTER_DOMAIN_RULES"])
async def test_avro_both_modes_validate_when_no_domain_rules_exist(mode):
    ser = await avro_serializer(**{'validation.rules.execution': mode})
    with pytest.raises(SerializationError, match="agePositive"):
        await ser({"age": -5, "name": "Alice"}, ser_ctx())

Copilot

Pull request overview

Copilot reviewed 28 out of 31 changed files in this pull request and generated no new comments.

Files not reviewed (3)

src/confluent_kafka/schema_registry/confluent/meta_pb2.py: Generated file
tests/schema_registry/data/proto/map_widget_pb2.py: Generated file
tests/schema_registry/data/proto/validation_widget_pb2.py: Generated file

Suppressed comments (7)

src/confluent_kafka/schema_registry/common/protobuf.py:363

With use.latest.version, descriptor is schema-side and may contain compatible fields that the generated runtime message does not yet have. Iterating those fields makes HasField/getattr raise when the latest schema added an optional field. Iterate message.DESCRIPTOR.fields and map each runtime field back to the schema descriptor so only available values are walked while rules still come from the latest schema.

    Walk ``message`` against ``descriptor``, evaluating every inline validation rule

src/confluent_kafka/schema_registry/common/json_schema.py:291

_validate_subtypes() temporarily mutates the supplied schema's type, but serializer instances cache and share this parsed dictionary across calls. Concurrent messages using a type array can therefore observe another call's temporary scalar type and skip the intended branch or its rules. Validate a shallow copy so the cached schema is never mutated.

    all_of = schema.get("allOf")

src/confluent_kafka/schema_registry/common/json_schema.py:312

When the matching oneOf branch is an external $ref, resolved contains only its contents while this call keeps the original resolver. Relative $refs inside that resource are then resolved against the root resource, unlike the direct $ref path below which correctly uses the lookup's scoped resolver. Recurse through the original branch so the normal $ref handling preserves resolver scope.

                        return

src/confluent_kafka/schema_registry/common/json_schema.py:318

The matching anyOf branch has the same resolver loss: an external $ref is replaced by its contents, then walked with the root resolver, so nested relative references can resolve to the wrong resource. Recurse through subschema and let the dedicated $ref path carry the scoped resolver returned by lookup.

        if fail_fast and out:

src/confluent_kafka/schema_registry/rules/cel/cel_validator.py:126

Avro/JSON boolean values are converted incorrectly here: the shared _value_to_cel() checks int before bool, and Python booleans are integers. Consequently this == true (including booleans nested in records/lists) is evaluated with an IntType and can become a rule-execution violation. Handle bool before int in the shared recursive converter and add scalar/nested boolean coverage.

    return _value_to_cel(value)

src/confluent_kafka/schema_registry/common/avro.py:356

This fallback ignores an explicit namespace on the candidate schema. For a union containing a.Rec and b.Rec, an invalid wrapper name such as c.Rec misses the exact pass and then incorrectly selects the first Rec. Restrict simple-name fallback to schemas whose namespace is genuinely unavailable.

    return '.' not in name and branch_name.rsplit('.', 1)[-1] == name

src/confluent_kafka/schema_registry/common/avro.py:277

Named Avro references are strings too, so this fallthrough does not only represent primitive leaves. In particular, fastavro.expand_schema() intentionally leaves recursive self-references as names to avoid an infinite schema, causing rules on a recursive record to run only at the root and not on nested values; ordinary parsed schemas also miss reused named records. Resolve non-primitive names through a named-schema table during the walk before treating them as leaves.

    # primitive leaf — field-level rules were evaluated by the parent record case

Copilot

Pull request overview

Copilot reviewed 28 out of 31 changed files in this pull request and generated no new comments.

Files not reviewed (3)

src/confluent_kafka/schema_registry/confluent/meta_pb2.py: Generated file
tests/schema_registry/data/proto/map_widget_pb2.py: Generated file
tests/schema_registry/data/proto/validation_widget_pb2.py: Generated file

Suppressed comments (2)

src/confluent_kafka/schema_registry/rules/cel/cel_validator.py:126

Avro/JSON boolean values are converted incorrectly here. _value_to_cel() checks int before bool, and Python booleans are integers, so a rule such as this == true receives IntType(1) rather than BoolType(true) (including booleans nested in records/lists). Reorder the shared converter's boolean branch ahead of int and add Avro/JSON boolean-rule coverage.

    return _value_to_cel(value)

src/confluent_kafka/schema_registry/common/avro.py:224

Named Avro references are treated as primitive leaves, so validation stops at recursive references. expand_schema() cannot infinitely expand a self-reference (for example next: ["null", "Node"]), leaving "Node" here; rules on the nested Node are therefore never evaluated, contrary to the recursive-walk contract. Carry a named-schema lookup through the walker and resolve string references before dispatching.

    if schema is None:
        return
    if isinstance(schema, list):

Copilot

Pull request overview

Copilot reviewed 28 out of 31 changed files in this pull request and generated 4 comments.

Files not reviewed (3)

src/confluent_kafka/schema_registry/confluent/meta_pb2.py: Generated file
tests/schema_registry/data/proto/map_widget_pb2.py: Generated file
tests/schema_registry/data/proto/validation_widget_pb2.py: Generated file

Suppressed comments (7)

src/confluent_kafka/schema_registry/rules/cel/cel_validator.py:100

This cache is unbounded. A long-lived serializer using evolving latest schemas can retain one compiled CEL program for every distinct expression indefinitely; the referenced JVM validator caps this cache at 1,000 entries. Use a bounded LRU cache (or expose a configurable limit) to prevent schema evolution from causing persistent memory growth.

    def __init__(self):
        self.lock = Lock()
        self.programs = {}

src/confluent_kafka/schema_registry/_sync/avro.py:265

These three public serializer options are absent from AvroSerializer's configuration-properties table, even though that table documents every other supported option. Add their accepted values/types, defaults, execution ordering, and the optional CEL dependency/override behavior so users can discover and configure the feature from the API docs.

        'validation.rules.execution': ValidationRulesExecution.DISABLED,
        'validation.rules.fail.fast': False,
        'validation.rules.executor': None,

src/confluent_kafka/schema_registry/_sync/json_schema.py:254

These three public serializer options are absent from JSONSerializer's configuration-properties table, even though that table documents every other supported option. Add their accepted values/types, defaults, execution ordering, and the optional CEL dependency/override behavior so users can discover and configure the feature from the API docs.

        'validation.rules.execution': ValidationRulesExecution.DISABLED,
        'validation.rules.fail.fast': False,
        'validation.rules.executor': None,

src/confluent_kafka/schema_registry/_sync/protobuf.py:257

These three public serializer options are absent from ProtobufSerializer's configuration-properties table, even though that table documents every other supported option. Add their accepted values/types, defaults, execution ordering, and the optional CEL dependency/override behavior so users can discover and configure the feature from the API docs.

        'validation.rules.execution': ValidationRulesExecution.DISABLED,
        'validation.rules.fail.fast': False,
        'validation.rules.executor': None,

src/confluent_kafka/schema_registry/_async/avro.py:269

These three public serializer options are absent from AsyncAvroSerializer's configuration-properties table, even though that table documents every other supported option. Add their accepted values/types, defaults, execution ordering, and the optional CEL dependency/override behavior so users can discover and configure the feature from the API docs.

        'validation.rules.execution': ValidationRulesExecution.DISABLED,
        'validation.rules.fail.fast': False,
        'validation.rules.executor': None,

src/confluent_kafka/schema_registry/_async/json_schema.py:256

These three public serializer options are absent from AsyncJSONSerializer's configuration-properties table, even though that table documents every other supported option. Add their accepted values/types, defaults, execution ordering, and the optional CEL dependency/override behavior so users can discover and configure the feature from the API docs.

        'validation.rules.execution': ValidationRulesExecution.DISABLED,
        'validation.rules.fail.fast': False,
        'validation.rules.executor': None,

src/confluent_kafka/schema_registry/_async/protobuf.py:259

These three public serializer options are absent from AsyncProtobufSerializer's configuration-properties table, even though that table documents every other supported option. Add their accepted values/types, defaults, execution ordering, and the optional CEL dependency/override behavior so users can discover and configure the feature from the API docs.

        'validation.rules.execution': ValidationRulesExecution.DISABLED,
        'validation.rules.fail.fast': False,
        'validation.rules.executor': None,

Copilot

Pull request overview

Copilot reviewed 28 out of 31 changed files in this pull request and generated 1 comment.

Files not reviewed (3)

src/confluent_kafka/schema_registry/confluent/meta_pb2.py: Generated file
tests/schema_registry/data/proto/map_widget_pb2.py: Generated file
tests/schema_registry/data/proto/validation_widget_pb2.py: Generated file

Suppressed comments (7)

src/confluent_kafka/schema_registry/common/avro.py:230

Named Avro branches are not resolved here. fastavro represents references (including recursive record references such as ['null', 'Node']) as strings, so _resolve_union can return 'Node'; the recursive call then falls through without evaluating that nested record's record- or field-level rules. Carry the named-schema registry through this walk and resolve string references before dispatching; serializer-side expand_schema cannot eliminate recursive references.

        subschema, submessage = _resolve_union(schema, message)
        if subschema is None or subschema == 'null':
            return
        _validate_message(executor, subschema, submessage, path, fail_fast, out)

src/confluent_kafka/schema_registry/common/protobuf.py:368

This process-global cache strongly retains every registered/runtime descriptor pair ever seen. Descriptor objects are created per parsed schema/serializer, so repeatedly creating serializers or loading evolving schema versions grows this dictionary indefinitely and also prevents those descriptor pools from being reclaimed. Bound the cache (for example with an LRU) or scope it to the serializer/parsed-schema cache lifecycle.

_SCHEMA_VIEW_NEEDED: Dict[Tuple[Descriptor, Descriptor], bool] = {}

src/confluent_kafka/schema_registry/common/serde.py:492

The docstring says malformed entries are ignored, but every dict is accepted without checking that name, doc, expr, and sql are strings. For example, {"expr": 1} reaches CelValidator._program() and leaks a parser TypeError instead of being ignored or collected as a validation violation. Reject entries whose present fields are not strings before constructing ValidationRule.

    for entry in prop_value:
        if isinstance(entry, dict):
            rules.append(
                ValidationRule(
                    entry.get('name'),
                    entry.get('doc'),
                    entry.get('expr'),
                    entry.get('sql'),
                )

src/confluent_kafka/schema_registry/_sync/serde.py:285

These three customer-facing serializer options are implemented but absent from the Configuration properties tables on AvroSerializer, JSONSerializer, and ProtobufSerializer, even though every other accepted option is documented there. Add their types, valid execution values, defaults, ordering semantics, and executor contract so users can discover and configure this feature.

    def configure_validation_rules(self, conf: dict) -> None:
        """
        Pop and validate the inline validation rule configs from ``conf``.

        When execution is not DISABLED the executor is resolved eagerly, so a missing
        dependency surfaces at serializer construction rather than at the first record.

src/confluent_kafka/schema_registry/_async/serde.py:285

The async serializers now accept these three public options, but their AsyncAvroSerializer, AsyncJSONSerializer, and AsyncProtobufSerializer Configuration properties tables do not mention them. Document the accepted values, defaults, phase ordering, and executor interface consistently with the synchronous serializers.

    def configure_validation_rules(self, conf: dict) -> None:
        """
        Pop and validate the inline validation rule configs from ``conf``.

        When execution is not DISABLED the executor is resolved eagerly, so a missing
        dependency surfaces at serializer construction rather than at the first record.

tests/schema_registry/_sync/test_validation_serdes.py:145

This only proves that both modes trigger validation when there are no domain rules; it does not test the defining BEFORE/AFTER ordering. Add a domain transform that changes a value from failing to passing (or vice versa), then assert that the two modes produce different outcomes. Without that, regressions that run both phases on the same side of _execute_rules are undetected.

@pytest.mark.parametrize("mode", ["BEFORE_DOMAIN_RULES", "AFTER_DOMAIN_RULES"])
def test_avro_both_modes_validate_when_no_domain_rules_exist(mode):
    ser = avro_serializer(**{'validation.rules.execution': mode})
    with pytest.raises(SerializationError, match="agePositive"):
        ser({"age": -5, "name": "Alice"}, ser_ctx())

tests/schema_registry/_async/test_validation_serdes.py:147

This case has no domain transformation, so it cannot verify the advertised BEFORE/AFTER phase ordering. Exercise a domain rule that changes whether the inline rule passes and assert opposite outcomes for the two modes; otherwise the duplicated async pipeline ordering remains untested.

@pytest.mark.parametrize("mode", ["BEFORE_DOMAIN_RULES", "AFTER_DOMAIN_RULES"])
async def test_avro_both_modes_validate_when_no_domain_rules_exist(mode):
    ser = await avro_serializer(**{'validation.rules.execution': mode})
    with pytest.raises(SerializationError, match="agePositive"):
        await ser({"age": -5, "name": "Alice"}, ser_ctx())

Copilot

Pull request overview

Copilot reviewed 29 out of 32 changed files in this pull request and generated no new comments.

Files not reviewed (3)

src/confluent_kafka/schema_registry/confluent/meta_pb2.py: Generated file
tests/schema_registry/data/proto/map_widget_pb2.py: Generated file
tests/schema_registry/data/proto/validation_widget_pb2.py: Generated file

Suppressed comments (4)

src/confluent_kafka/schema_registry/rules/cel/cel_validator.py:126

Avro/JSON boolean values are converted incorrectly here because _value_to_cel() checks int before bool (and Python booleans are integers). A rule such as this == true, including a boolean nested in an object, therefore receives IntType(1/0) instead of BoolType and can reject valid data. Fix the shared recursive converter to handle bool before int, and cover scalar and object-level boolean rules.

    return _value_to_cel(value)

src/confluent_kafka/schema_registry/common/serde.py:521

This accepts any dictionary without validating the value types, despite the docstring saying malformed entries are ignored. For example, {"expr": 42} reaches CelValidator._program() with an integer and leaks a raw compiler/type exception instead of being ignored. Only construct a rule when name, doc, expr, and sql are each strings or None.

        if isinstance(entry, dict):
            rules.append(
                ValidationRule(
                    entry.get('name'),
                    entry.get('doc'),
                    entry.get('expr'),
                    entry.get('sql'),
                )

src/confluent_kafka/schema_registry/_async/serde.py:285

The async serializers also expose these three new configuration keys without adding them to the Avro, JSON Schema, or Protobuf serializer configuration tables. Document the execution modes, fail-fast default, and executor type in each async serializer's public docstring, consistent with every other supported option listed there.

    def configure_validation_rules(self, conf: dict) -> None:
        """
        Pop and validate the inline validation rule configs from ``conf``.

        When execution is not DISABLED the executor is resolved eagerly, so a missing
        dependency surfaces at serializer construction rather than at the first record.

src/confluent_kafka/schema_registry/_sync/serde.py:285

These are new customer-facing serializer options, but none of the sync Avro, JSON Schema, or Protobuf serializer configuration tables document their accepted values, defaults, or executor contract, while those tables document the other supported options. Add all three keys to the public serializer docstrings so users can discover and configure this feature.

    def configure_validation_rules(self, conf: dict) -> None:
        """
        Pop and validate the inline validation rule configs from ``conf``.

        When execution is not DISABLED the executor is resolved eagerly, so a missing
        dependency surfaces at serializer construction rather than at the first record.

Copilot

Pull request overview

Copilot reviewed 27 out of 30 changed files in this pull request and generated no new comments.

Files not reviewed (3)

src/confluent_kafka/schema_registry/confluent/meta_pb2.py: Generated file
tests/schema_registry/data/proto/map_widget_pb2.py: Generated file
tests/schema_registry/data/proto/validation_widget_pb2.py: Generated file

Suppressed comments (8)

src/confluent_kafka/schema_registry/common/avro.py:270

Skip rules only for a nullable union whose runtime value is null. As written, every missing/null field is skipped, including non-nullable fields; this diverges from the linked JVM implementation and can let an omitted field that fastavro serializes from a failing default bypass its CHECK rule. Resolve an absent field to its Avro default and still invoke the executor for non-nullable nulls.

                value = message.get(name)
                # Skip-on-null: an absent or null field value does not invoke the
                # executor. The recursion below still runs but no-ops for None.
                if value is not None:
                    for rule in _read_validation_rules(field):
                        evaluate_validation_rule(executor, rule, field_schema, value, child_path, out)

src/confluent_kafka/schema_registry/common/protobuf.py:504

This equivalence check ignores declared defaults. For two proto2 descriptors with the same field shape but different defaults, _needs_schema_view() returns false, so a message-level CEL rule reading an unset field observes the producer class's default instead of the registered schema's default and can return the wrong result. Treat differing defaults as requiring a schema view.

        if (
            schema_fd.name != runtime_fd.name
            or schema_fd.type != runtime_fd.type
            or _is_repeated(schema_fd) != _is_repeated(runtime_fd)
        ):

src/confluent_kafka/schema_registry/_sync/avro.py:265

Add these three public options to AvroSerializer's “Configuration properties” table. Every other accepted option is documented there, but users currently cannot discover the validation modes, executor type, or defaults from the API reference.

        'validation.rules.execution': ValidationRulesExecution.DISABLED,
        'validation.rules.fail.fast': False,
        'validation.rules.executor': None,

src/confluent_kafka/schema_registry/_sync/json_schema.py:254

Add these three public options to JSONSerializer's “Configuration properties” table. Every other accepted option is documented there, but users currently cannot discover the validation modes, executor type, or defaults from the API reference.

        'validation.rules.execution': ValidationRulesExecution.DISABLED,
        'validation.rules.fail.fast': False,
        'validation.rules.executor': None,

src/confluent_kafka/schema_registry/_sync/protobuf.py:257

Add these three public options to ProtobufSerializer's “Configuration properties” table. Every other accepted option is documented there, but users currently cannot discover the validation modes, executor type, or defaults from the API reference.

        'validation.rules.execution': ValidationRulesExecution.DISABLED,
        'validation.rules.fail.fast': False,
        'validation.rules.executor': None,

src/confluent_kafka/schema_registry/_async/avro.py:269

Add these three public options to AsyncAvroSerializer's “Configuration properties” table. Every other accepted option is documented there, but users currently cannot discover the validation modes, executor type, or defaults from the API reference.

        'validation.rules.execution': ValidationRulesExecution.DISABLED,
        'validation.rules.fail.fast': False,
        'validation.rules.executor': None,

src/confluent_kafka/schema_registry/_async/json_schema.py:256

Add these three public options to AsyncJSONSerializer's “Configuration properties” table. Every other accepted option is documented there, but users currently cannot discover the validation modes, executor type, or defaults from the API reference.

        'validation.rules.execution': ValidationRulesExecution.DISABLED,
        'validation.rules.fail.fast': False,
        'validation.rules.executor': None,

src/confluent_kafka/schema_registry/_async/protobuf.py:259

Add these three public options to AsyncProtobufSerializer's “Configuration properties” table. Every other accepted option is documented there, but users currently cannot discover the validation modes, executor type, or defaults from the API reference.

        'validation.rules.execution': ValidationRulesExecution.DISABLED,
        'validation.rules.fail.fast': False,
        'validation.rules.executor': None,

Copilot

Pull request overview

Copilot reviewed 28 out of 31 changed files in this pull request and generated 1 comment.

Files not reviewed (3)

src/confluent_kafka/schema_registry/confluent/meta_pb2.py: Generated file
tests/schema_registry/data/proto/map_widget_pb2.py: Generated file
tests/schema_registry/data/proto/validation_widget_pb2.py: Generated file

Suppressed comments (7)

src/confluent_kafka/schema_registry/common/protobuf.py:370

This process-global cache strongly retains every schema/runtime descriptor pair. Each parsed schema version produces distinct descriptor objects, so applications that create serializers or observe evolving latest schemas accumulate entries indefinitely, and discarded descriptor pools cannot be collected. Use a bounded or weak cache, or scope it to the serializer/parsed-schema cache.

_SCHEMA_VIEW_NEEDED: Dict[Tuple[Descriptor, Descriptor], bool] = {}

src/confluent_kafka/schema_registry/common/protobuf.py:500

The equality check omits proto2 field defaults. If the registered schema changes an unset scalar's default while retaining its name/type/label, this returns “same,” so message-level rules read the producer descriptor's old default instead of the registered schema's default. Compare defaults too, taking care to compare enum defaults by value/number rather than descriptor identity.

        if (
            schema_fd.name != runtime_fd.name
            or schema_fd.type != runtime_fd.type
            or _is_repeated(schema_fd) != _is_repeated(runtime_fd)
        ):

src/confluent_kafka/schema_registry/_sync/avro.py:500

With record_subject_name_strategy, callers may omit SerializationContext while still resolving a subject and latest_schema. The enclosing latest-schema branch then falls through here, and validation uses self._parsed_schema, silently ignoring inline rules from the selected latest schema. Resolve the validation schema from latest_schema independently of whether domain-rule execution has a context.

            if self._validation_enabled():
                self._validate_inline_rules(expand_schema(parsed_schema), value)

src/confluent_kafka/schema_registry/_async/avro.py:504

With record_subject_name_strategy, callers may omit SerializationContext while still resolving a subject and latest_schema. The enclosing latest-schema branch then falls through here, and validation uses self._parsed_schema, silently ignoring inline rules from the selected latest schema. Resolve the validation schema from latest_schema independently of whether domain-rule execution has a context.

            if self._validation_enabled():
                self._validate_inline_rules(expand_schema(parsed_schema), value)

src/confluent_kafka/schema_registry/_sync/avro.py:481

The new tests exercise both execution modes only when no domain rules exist, so they cannot detect whether validation was actually placed before versus after transformations. Add a serializer-level domain transform that changes a message from invalid to valid and assert that BEFORE_DOMAIN_RULES fails while AFTER_DOMAIN_RULES succeeds (including async parity).

            if self._validation_enabled(ValidationRulesExecution.BEFORE_DOMAIN_RULES):
                self._validate_inline_rules(expanded_parsed_schema, value)

src/confluent_kafka/schema_registry/_sync/serde.py:287

These are customer-facing serializer options, but the Avro, JSON Schema, and Protobuf serializer “Configuration properties” tables do not list them. As a result, generated API documentation does not expose accepted execution values, defaults, fail-fast behavior, or the executor contract. Document all three options in the sync and async serializer class tables.

        execution = conf.pop('validation.rules.execution', ValidationRulesExecution.DISABLED)

src/confluent_kafka/schema_registry/common/json_schema.py:304

_validate_subschema resolves a $ref but returns only its contents, discarding the resolver scoped to that external resource. Recursing here with the original resolver makes any relative $ref inside a selected oneOf/anyOf branch resolve against the wrong base URI (or fail), even though the direct $ref path below correctly propagates ref_schema.resolver. Preserve and pass the resolver returned by lookup.

                resolved = _validate_subschema(subschema, message, ref_registry, ref_resolver)
                if resolved is not None:
                    _validate_message(executor, resolved, ref_registry, ref_resolver, path, message, fail_fast, out)

Copilot

Pull request overview

Copilot reviewed 28 out of 31 changed files in this pull request and generated no new comments.

Files not reviewed (3)

src/confluent_kafka/schema_registry/confluent/meta_pb2.py: Generated file
tests/schema_registry/data/proto/map_widget_pb2.py: Generated file
tests/schema_registry/data/proto/validation_widget_pb2.py: Generated file

Suppressed comments (3)

src/confluent_kafka/schema_registry/common/json_schema.py:310

When a oneOf/anyOf candidate is an external $ref, _validate_subschema returns the referenced contents but discards the scoped resolver. Walking those contents with the caller's resolver makes any relative $ref inside the external resource resolve against the wrong base URI (or fail). After using _validate_subschema only to select matching candidates, recurse into the original candidate so the normal $ref branch can retain lookup(...).resolver. Apply the same change to anyOf.

                resolved = _validate_subschema(subschema, message, ref_registry, ref_resolver)
                if resolved is not None:
                    _validate_message(executor, resolved, ref_registry, ref_resolver, path, message, fail_fast, out)

src/confluent_kafka/schema_registry/common/protobuf.py:500

The equality check misses the descriptor's map_entry shape. A protobuf map is wire-compatible with its repeated entry-message representation, so latest and runtime schemas can agree on name/type/label while one exposes a CEL map and the other a list. This currently skips the schema re-read and evaluates message-level rules against the wrong value shape. Include map-ness in the comparison.

        if (
            schema_fd.name != runtime_fd.name
            or schema_fd.type != runtime_fd.type
            or _is_repeated(schema_fd) != _is_repeated(runtime_fd)
        ):

src/confluent_kafka/schema_registry/_sync/serde.py:285

These three customer-facing options are absent from the “Configuration properties” tables on all six sync/async Avro, JSON Schema, and Protobuf serializer classes, even though those tables otherwise enumerate supported options. Add entries documenting the valid execution modes/default, fail-fast behavior, executor type, and optional CEL dependency so users can configure this feature from the API docs.

    def configure_validation_rules(self, conf: dict) -> None:
        """
        Pop and validate the inline validation rule configs from ``conf``.

        When execution is not DISABLED the executor is resolved eagerly, so a missing
        dependency surfaces at serializer construction rather than at the first record.

Copilot

Pull request overview

Copilot reviewed 28 out of 31 changed files in this pull request and generated 1 comment.

Files not reviewed (3)

src/confluent_kafka/schema_registry/confluent/meta_pb2.py: Generated file
tests/schema_registry/data/proto/map_widget_pb2.py: Generated file
tests/schema_registry/data/proto/validation_widget_pb2.py: Generated file

Suppressed comments (7)

src/confluent_kafka/schema_registry/common/json_schema.py:386

Object-valued property rules are evaluated twice: this loop evaluates prop_schema and then _validate_message() recurses into the same object schema, where _validate_object() evaluates the same rules again. A single failing rule on an object property is consequently reported as two violations. Evaluate rules once at the start of _validate_message() (as the JVM walker does) and let this property loop only recurse.

        if value is not None:
            for rule in _read_validation_rules(prop_schema):
                evaluate_validation_rule(executor, rule, prop_schema, value, full_name, out)
                if fail_fast and out:
                    return
        _validate_message(executor, prop_schema, ref_registry, ref_resolver, full_name, value, fail_fast, out)

src/confluent_kafka/schema_registry/_sync/avro.py:265

These new public options are absent from AvroSerializer's configuration table above, although that table documents the other supported entries in _default_conf. Add their accepted values, defaults, and execution semantics so users can discover and configure inline validation from the generated API documentation.

        'validation.rules.execution': ValidationRulesExecution.DISABLED,
        'validation.rules.fail.fast': False,
        'validation.rules.executor': None,

src/confluent_kafka/schema_registry/_sync/json_schema.py:254

These new public options are absent from JSONSerializer's configuration table above, although that table documents the other supported entries in _default_conf. Add their accepted values, defaults, and execution semantics so users can discover and configure inline validation from the generated API documentation.

        'validation.rules.execution': ValidationRulesExecution.DISABLED,
        'validation.rules.fail.fast': False,
        'validation.rules.executor': None,

src/confluent_kafka/schema_registry/_sync/protobuf.py:257

These new public options are absent from ProtobufSerializer's configuration table above, although that table documents the other supported entries in _default_conf. Add their accepted values, defaults, and execution semantics so users can discover and configure inline validation from the generated API documentation.

        'validation.rules.execution': ValidationRulesExecution.DISABLED,
        'validation.rules.fail.fast': False,
        'validation.rules.executor': None,

src/confluent_kafka/schema_registry/_async/avro.py:269

These new public options are absent from AsyncAvroSerializer's configuration table above, although that table documents the other supported entries in _default_conf. Add their accepted values, defaults, and execution semantics so async users can discover and configure inline validation from the generated API documentation.

        'validation.rules.execution': ValidationRulesExecution.DISABLED,
        'validation.rules.fail.fast': False,
        'validation.rules.executor': None,

src/confluent_kafka/schema_registry/_async/json_schema.py:256

These new public options are absent from AsyncJSONSerializer's configuration table above, although that table documents the other supported entries in _default_conf. Add their accepted values, defaults, and execution semantics so async users can discover and configure inline validation from the generated API documentation.

        'validation.rules.execution': ValidationRulesExecution.DISABLED,
        'validation.rules.fail.fast': False,
        'validation.rules.executor': None,

src/confluent_kafka/schema_registry/_async/protobuf.py:259

These new public options are absent from AsyncProtobufSerializer's configuration table above, although that table documents the other supported entries in _default_conf. Add their accepted values, defaults, and execution semantics so async users can discover and configure inline validation from the generated API documentation.

        'validation.rules.execution': ValidationRulesExecution.DISABLED,
        'validation.rules.fail.fast': False,
        'validation.rules.executor': None,

Copilot

Pull request overview

Copilot reviewed 28 out of 31 changed files in this pull request and generated no new comments.

Files not reviewed (3)

src/confluent_kafka/schema_registry/confluent/meta_pb2.py: Generated file
tests/schema_registry/data/proto/map_widget_pb2.py: Generated file
tests/schema_registry/data/proto/validation_widget_pb2.py: Generated file

Suppressed comments (3)

src/confluent_kafka/schema_registry/common/json_schema.py:304

The selected oneOf/anyOf branch may have been resolved from an external resource, but recursion switches back to the original resolver. _validate_subschema() validates with the lookup-scoped resolver and then returns only contents; if that content contains a relative or local $ref, this call resolves it against the root resource, so nested inline rules are missed or lookup fails. Preserve and pass the resolver returned by the lookup together with the resolved schema, as the direct $ref branch below already does.

                resolved = _validate_subschema(subschema, message, ref_registry, ref_resolver)
                if resolved is not None:
                    _validate_message(executor, resolved, ref_registry, ref_resolver, path, message, fail_fast, out)

src/confluent_kafka/schema_registry/common/avro.py:364

Stripping an arbitrary namespace here can select the wrong Avro union branch. For example, inside namespace a, an inherited branch named Rec has fullname a.Rec, but a wrapped value naming b.Rec also matches this fallback. The enclosing namespace is not available in this function, so it cannot safely infer that any qualified name with the same suffix refers to this branch; resolve/normalize the branch fullname using the enclosing schema context instead.

    return '.' not in name and not subschema.get("namespace") and branch_name.rsplit('.', 1)[-1] == name

src/confluent_kafka/schema_registry/_sync/serde.py:301

These three new customer-facing serializer options are absent from the configuration-property tables for AvroSerializer, JSONSerializer, and ProtobufSerializer (and their async counterparts). Users therefore cannot discover the allowed execution values, defaults, fail-fast behavior, or executor contract from the API documentation. Add the options to all six public serializer docstrings.

        execution = conf.pop('validation.rules.execution', ValidationRulesExecution.DISABLED)
        try:
            self._validation_rules_execution = ValidationRulesExecution(execution)
        except ValueError:
            raise ValueError(
                "validation.rules.execution must be one of {}".format(
                    ", ".join(m.value for m in ValidationRulesExecution)
                )
            )

        self._validation_rules_fail_fast = cast(bool, conf.pop('validation.rules.fail.fast', False))
        if not isinstance(self._validation_rules_fail_fast, bool):
            raise ValueError("validation.rules.fail.fast must be a boolean value")

        executor = conf.pop('validation.rules.executor', None)

sonarqube-confluent · 2026-08-15T01:33:33Z

Quality Gate passed

Issues
16 New issues
0 Fixed issues
0 Accepted issues

Measures
0 Security Hotspots
84.6% Coverage on New Code
7.1% Duplication on New Code

See analysis details on SonarQube

Robert Yokota (rayokota) requested review from a team and Matthew Seal (MSeal) as code owners August 11, 2026 23:11

Copilot AI lite review requested due to automatic review settings August 11, 2026 23:11

Robert Yokota (rayokota) requested a review from a team as a code owner August 11, 2026 23:11

Copilot started reviewing on behalf of Robert Yokota (rayokota) August 11, 2026 23:11 View session

Copilot AI reviewed Aug 11, 2026

View reviewed changes

Comment thread src/confluent_kafka/schema_registry/common/serde.py Outdated

Comment thread src/confluent_kafka/schema_registry/common/protobuf.py Outdated

Robert Yokota (rayokota) requested a balanced review from Copilot August 12, 2026 20:50

Copilot started reviewing on behalf of Robert Yokota (rayokota) August 12, 2026 20:51 View session

Copilot AI reviewed Aug 12, 2026

View reviewed changes

Comment thread src/confluent_kafka/schema_registry/common/json_schema.py Outdated

Robert Yokota (rayokota) requested a balanced review from Copilot August 12, 2026 21:47

Copilot started reviewing on behalf of Robert Yokota (rayokota) August 12, 2026 21:47 View session

Copilot AI reviewed Aug 12, 2026

View reviewed changes

Comment thread src/confluent_kafka/schema_registry/common/protobuf.py Outdated

Comment thread src/confluent_kafka/schema_registry/common/json_schema.py Outdated

Robert Yokota (rayokota) requested a balanced review from Copilot August 13, 2026 00:22

Copilot started reviewing on behalf of Robert Yokota (rayokota) August 13, 2026 00:22 View session