Add Variant, Decimal, and Timestamp CEL functions

[Robert Yokota (rayokota)]

What

Summary

Adds three families of CEL functions to data contract rules, bringing this client to parity
with the JVM reference implementation:

• variant(...) / variants.* — read and navigate a Spark/Parquet Variant
• decimal(...) / decimals.* — exact decimal arithmetic and comparison
• timestamp(...) extensions — construct from an epoch value at a given precision, and
  accept the temporal shapes the Avro and Protobuf decoders produce

Before this, a rule could not work with any of these three types: a confluent.type.Decimal
or google.protobuf.Timestamp field reached CEL as an opaque message, an Avro decimal reached
it as raw unscaled bytes, and a Variant was unreachable entirely.

What's added

Variant — variant(dyn) and variant(value, metadata) constructors; variants.parseJson
(strict) and variants.tryParseJson (CEL null on a malformed document); variants.type;
navigation via variants.field, variants.index and variants.path (a JSONPath subset:
$, $.field, $[i], $["quoted key"]); typed extraction via variants.as / variants.tryAs;
plus variants.isNull and variants.toJson.

Decimal — decimal(...) from a string, int, uint, double or unscaled-bytes-plus-scale;
arithmetic (add, sub, mul, div, mod); rounding (round, trunc, floor, ceil);
abs and sign; comparisons; and string(...) / double(...) extended to accept a Decimal.

Timestamp — timestamp(value, precision) where precision is one of {0, 3, 6, 9}
(seconds, millis, micros, nanos), and a timestamp(dyn) overload accepting the temporal
representations a decoder hands back. string(...) renders a timestamp with its sub-second
component.

Marshalling boundary — the schema-side value is converted to its CEL type on the way in
and back to the schema's representation on the way out, for both field-level (CEL_FIELD) and
message-level (CEL) rules, across Avro, Protobuf and JSON Schema. A decimal keeps its scale,
a timestamp keeps its unit, and a Variant round-trips as a Variant.

Semantics

The JVM client is the contract; behaviour here is matched against it rather than against this
language's native conventions. In particular:

• Exact arithmetic. add, sub, mul and mod are exact, as java.math.BigDecimal is.
  Division is capped at 38 significant digits with HALF_UP, matching the JVM's DIV_MC.
• Scale is part of the value. 12.34 and 12.340 are the same number in two encodings and
  are rendered differently; round/trunc produce exactly the requested scale, including a
  negative one (round(1234, -2) is 1200). Scale arguments are int32-bounded, as
  BigDecimal's are.
• Wire form. The unscaled value is minimal big-endian two's complement, byte-identical to
  BigInteger.toByteArray(), and precision is the unscaled value's digit count as
  BigDecimal.precision() reports it.
• Range and type checks are errors, not coercions. A non-finite double, a timestamp outside
  0001-01-01T00:00:00Z .. 9999-12-31T23:59:59.999999999Z, an out-of-range scale, or a
  wrong-typed argument is a rule error rather than something silently narrowed — the JVM's
  typed overloads reject the same inputs.

Known limitations

These are deliberate and shared across the non-JVM clients:

• float/double JSON rendering stays native to this language. Byte-identical rendering
  across all clients was designed and implemented, then backed out: the precision walk it
  requires costs 14–23× a native format call and about 75% of the serialization path, and no
  cross-client bug had been reported against it. Values are equal; their shortest-form text may
  differ.
• precision is informational on read. The JVM applies it as a MathContext when decoding;
  this client returns the value unrounded. Since every client now writes precision as the
  value's own digit count, the two agree for anything these clients produce.
• Avro local-timestamp-* is not converted. It carries no zone, so the JVM refuses to turn
  it into an instant; conversion support here is tracked separately.

Checklist

• Contains customer facing changes? Including API/behavior changes
• 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]

🎉 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 CEL support for Variant, Decimal, and Timestamp values, with Avro/Protobuf integration and serializer and validator tests.

Changes:

• Adds Variant codecs, builders, JSON conversion, and path navigation.
• Adds Decimal conversions/operators and Timestamp overloads.
• Extends CEL dispatch and serialization integrations.
• Adds unit and synchronous/asynchronous integration tests.

Reviewed changes

Copilot reviewed 19 out of 20 changed files in this pull request and generated 14 comments.

Show a summary per file

File	Summary
tests/schema_registry/test_variant_utils.py	Variant codec and timestamp tests.
tests/schema_registry/test_cel_validator.py	CEL behavior and integration tests.
tests/schema_registry/_sync/test_proto_serdes.py	Synchronous Protobuf integration tests.
tests/schema_registry/_sync/test_avro_serdes.py	Synchronous Avro integration tests.
tests/schema_registry/_async/test_proto_serdes.py	Asynchronous Protobuf integration tests.
tests/schema_registry/_async/test_avro_serdes.py	Asynchronous Avro integration tests.
src/confluent_kafka/schema_registry/rules/cel/variant_path.py	Variant path parsing. Nit (2 votes): identifier checks accept Unicode instead of the documented ASCII grammar.
src/confluent_kafka/schema_registry/rules/cel/variant_funcs.py	Variant CEL functions. Moderate (2 votes): tryParseJson accepts non-string inputs. Moderate (3 votes): index conversion truncates doubles and accepts booleans.
src/confluent_kafka/schema_registry/rules/cel/timestamp_funcs.py	Timestamp CEL overloads. Moderate (4 votes): two-argument overflow escapes as a raw exception. Moderate (3 votes): naive timestamps can bypass rejection.
src/confluent_kafka/schema_registry/rules/cel/extra_func.py	Registers extended CEL functions.
src/confluent_kafka/schema_registry/rules/cel/decimal_funcs.py	Decimal CEL functions. Moderate (2 votes): BoolType is treated as an integer. Moderate (2 votes): nested Decimal message wrappers are not converted correctly.
src/confluent_kafka/schema_registry/rules/cel/cel_validator.py	CEL validation and Decimal boundary conversion.
src/confluent_kafka/schema_registry/rules/cel/cel_field_presence.py	Namespaced CEL dispatch.
src/confluent_kafka/schema_registry/rules/cel/cel_executor.py	CEL value conversion and lazy now binding.
src/confluent_kafka/schema_registry/confluent/types/variant.proto	Variant Protobuf schema.
src/confluent_kafka/schema_registry/confluent/types/variant_utils.py	Variant codec and builder. Moderate (3 votes): truncated decimals raise IndexError. Moderate (2 votes): integer capacity checks reserve too much space. Moderate (4 votes): negative zero loses its sign in JSON. Moderate (2 votes): decimal capacity checks reject values that fit their selected width.
src/confluent_kafka/schema_registry/confluent/types/variant_pb2.py	Generated Variant Protobuf bindings.
src/confluent_kafka/schema_registry/confluent/types/decimal_utils.py	Decimal Protobuf conversions. Moderate (4 votes): ambient precision can round large values. Moderate (4 votes): negative boundary values produce non-canonical bytes.
src/confluent_kafka/schema_registry/common/protobuf.py	Variant Protobuf integration.
src/confluent_kafka/schema_registry/common/avro.py	Avro Variant logical-type integration. Critical (1 vote): logical handlers are registered through incorrect fastavro objects, preventing Variant round-tripping.

Files not reviewed (1)

• src/confluent_kafka/schema_registry/confluent/types/variant_pb2.py: Generated file

Suppressed comments (12)

src/confluent_kafka/schema_registry/confluent/types/variant_utils.py:1100

• The builder accepts a caller-supplied size_limit, but _integer_size() still returns a three-byte width for values above 0xFFFFFF. Any container or metadata larger than that then fails in to_bytes(3) with a raw OverflowError even though the configured limit permits it. Return a four-byte width after the 24-bit range (the header already supports four widths).

def _integer_size(value: int) -> int:
    if value <= U8_MAX:
        return 1
    if value <= U16_MAX:
        return 2
    return U24_SIZE

src/confluent_kafka/schema_registry/confluent/types/variant_utils.py:274

• Java Float.toString(-0.0f) also preserves the sign, but this branch formats it with int(f) as 0.0. The resulting Variant JSON differs from the documented Java contract; exclude zero from the integer branch so the existing repr() path retains -0.0.

    if f == int(f) and abs(f) < 1e16:
        return "%d.0" % int(f)

src/confluent_kafka/schema_registry/confluent/types/variant_utils.py:278

• The float formatter claims to match Java Float.toString, but repr(float(s)) uses Python's exponent formatting. For example, a stored float32 value around 1e-7 renders as 1e-07, whereas Java renders 1.0E-7; exact to_json() comparisons therefore diverge for scientific-notation values. Use a formatter with Java's exponent thresholds/casing and required mantissa digit instead of returning Python repr() directly.

    for p in range(1, 10):
        s = "%.*g" % (p, f)
        if struct.unpack("<f", struct.pack("<f", float(s)))[0] == f:
            return repr(float(s))

src/confluent_kafka/schema_registry/confluent/types/variant_utils.py:186

• Metadata field names are part of the Variant UTF-8 contract, but a malformed byte sequence raises UnicodeDecodeError directly here rather than VariantError. For a raw/protobuf Variant this escapes the CEL function boundary as an unhandled Python exception; normalize invalid UTF-8 to the codec's documented malformed-input error.

    return metadata[string_start + offset:string_start + next_offset].decode("utf-8")

src/confluent_kafka/schema_registry/confluent/types/variant_utils.py:465

• A malformed UTF-8 string payload also raises UnicodeDecodeError directly from get_string(), despite VariantError being the reader's documented malformed-input exception. This is especially visible through variants.as(..., 'string'), where the raw exception bypasses CEL error handling; catch the decode error and raise VariantError.

        return self.value[start:start + length].decode("utf-8")

src/confluent_kafka/schema_registry/confluent/types/variant_utils.py:538

• Negative field indexes are not validated here, so Python's negative indexing returns the last object field instead of rejecting the index. get_element_at_index explicitly rejects negative indexes, and JSONPath declares the same non-negative rule; validate the field index before indexing the encoded tables.

        key_id, value_pos = self._field_id_and_offset(idx)

src/confluent_kafka/schema_registry/rules/cel/cel_field_presence.py:139

• The namespace override calls func(*args) directly, bypassing celpy's normal conversion of function exceptions into CELEvalError. The new variants.* functions can raise VariantError/IndexError from malformed wire data (for example, a proto Variant with invalid metadata), so CelValidator.execute then leaks the raw exception instead of raising its documented RuleError; preserve existing CELEvalError and normalize other runtime exceptions at this dispatch boundary.

                                return func(*args)

src/confluent_kafka/schema_registry/rules/cel/decimal_funcs.py:382

• As with _string, this arm only handles a raw Decimal. A selected protobuf decimal field is a celpy MessageType wrapper, so double(this.decimal_field) falls through to DoubleType with a mapping and raises instead of performing the documented decimal-to-double conversion. Reuse decimal_boundary_value() before delegating.

    if isinstance(v, Decimal):
        return celtypes.DoubleType(float(v))
    return _STDLIB_DOUBLE(v)

src/confluent_kafka/schema_registry/rules/cel/decimal_funcs.py:62

• The (bytes, scale) overload is declared with an integer scale, but int(scale) silently truncates doubles and accepts CEL booleans (2.9 becomes scale 2, true becomes 1). This can produce a valid but unintended decimal instead of reporting an invalid overload argument; validate the CEL integer type before conversion.

def _from_bytes_scale(value: typing.Any, scale: typing.Any) -> Decimal:
    """Construct a Decimal from raw two's-complement big-endian bytes + scale."""
    raw = _coerce_bytes(value)
    s = int(scale)
    if len(raw) == 0:
        return Decimal(0).scaleb(-s, context=_EXACT_CONTEXT)
    return Decimal(int.from_bytes(raw, "big", signed=True)).scaleb(-s, context=_EXACT_CONTEXT)

src/confluent_kafka/schema_registry/rules/cel/decimal_funcs.py:296

• The target scale is documented as an integer, but int(args[1]) silently truncates a CEL double (for example, decimals.round(d, 1.9) rounds at scale 1) and accepts booleans. Validate the CEL integer type rather than coercing arbitrary values; the same validation should be shared with the other scale-taking overloads.

def _decimals_round(*args: typing.Any) -> Decimal:
    """Round to the given scale (HALF_UP). One-arg form rounds to integer."""
    if len(args) == 1:
        return _d(args[0]).quantize(
            Decimal(1), rounding=decimal.ROUND_HALF_UP, context=_EXACT_CONTEXT)
    if len(args) == 2:
        scale = int(args[1])
        return _d(args[0]).quantize(
            Decimal(1).scaleb(-scale), rounding=decimal.ROUND_HALF_UP,
            context=_EXACT_CONTEXT)

src/confluent_kafka/schema_registry/rules/cel/decimal_funcs.py:324

• As in decimals.round, int(args[1]) silently truncates a non-integer CEL value and accepts booleans even though this overload requires an integer target scale. This can truncate at a scale different from the caller's value; reuse the shared integer-scale validation before conversion.

    if len(args) == 2:
        d = _d(args[0])
        scale = int(args[1])
        if scale >= -d.as_tuple().exponent:
            return d
        return d.quantize(
            Decimal(1).scaleb(-scale), rounding=decimal.ROUND_DOWN,
            context=_EXACT_CONTEXT)

src/confluent_kafka/schema_registry/rules/cel/variant_funcs.py:230

• variants.field is documented with a string key, but str(key) accepts arbitrary CEL values. On an object containing a numeric-looking key, variants.field(v, 1) can silently access "1" instead of reporting a bad argument type, unlike the strict parseJson overload. Validate str/StringType before coercing the key.

    return v.get_field_by_key(str(key))

---

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

[sonarqube-confluent]

Quality Gate failed

Failed conditions
76.4% Coverage on New Code (required ≥ 80%)

See analysis details on SonarQube

[Copilot]

🟡 Changes recommended

Repeated value-type conditions, Variant path validation, and wide Decimal precision handling contain correctness defects.

Once you've addressed the issues Copilot identified, you can request another Copilot review.

Review details

Files not reviewed (3)

• src/confluent_kafka/schema_registry/confluent/types/variant_pb2.py: Generated file
• tests/schema_registry/data/proto/value_type_rules_pb2.py: Generated file
• tests/schema_registry/data/proto/value_types_pb2.py: Generated file

Suppressed comments (1)

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

• Repeated Decimal/Timestamp conditions are silently accepted when any element fails. This new leaf path makes each list element return a boolean, but the parent receives a list such as [False, True], and _transform_field only rejects the singleton False. Aggregate condition results before returning the list so a failing element propagates to the field.

    if isinstance(message, Message) and not is_cel_leaf_message(message.DESCRIPTOR):

• Files reviewed: 31/34 changed files
• Comments generated: 2
• Review effort level: Balanced

[Copilot]

🟡 Changes recommended

CEL uint handling and Decimal/Timestamp parity gaps allow behavior that conflicts with the declared JVM-compatible API.

Once you've addressed the issues Copilot identified, you can request another Copilot review.

Review details

Files not reviewed (3)

• src/confluent_kafka/schema_registry/confluent/types/variant_pb2.py: Generated file
• tests/schema_registry/data/proto/value_type_rules_pb2.py: Generated file
• tests/schema_registry/data/proto/value_types_pb2.py: Generated file

Suppressed comments (3)

src/confluent_kafka/schema_registry/rules/cel/decimal_funcs.py:200

• Python's str(float) is not equivalent to Java's Double.toString, which BigDecimal.valueOf uses. This changes observable scale: for 10000000.0, Python builds Decimal("10000000.0") (so string(decimal(...)) keeps .0), while the JVM parses 1.0E7 and renders 10000000; 1e20 likewise produces a different protobuf scale. Use a Java-compatible double-to-decimal canonicalization rather than Python str so the documented JVM/scale contract holds.

    if isinstance(v, float):
        # Java uses BigDecimal.valueOf(double), which throws on NaN/Infinity.
        # str() of a non-finite float ("nan"/"inf"/"-inf") builds a poisoned
        # Decimal in Python, so validate through the same finite check.
        return _decimal_from_string(str(v), v)

src/confluent_kafka/schema_registry/rules/cel/timestamp_funcs.py:171

• These checks accept celtypes.UintType because it subclasses int, allowing both timestamp(1u, 0) and timestamp(1, 0u) despite this overload being (int, int). Explicitly reject uint values so wrong-typed calls fail instead of being silently reinterpreted.

        if isinstance(value, (bool, celtypes.BoolType)) or not isinstance(value, (int, celtypes.IntType)):
            raise celpy.CELEvalError(f"timestamp: epoch value must be int, got {type(value).__name__}")
        if isinstance(precision, (bool, celtypes.BoolType)) or not isinstance(precision, (int, celtypes.IntType)):
            raise celpy.CELEvalError(f"timestamp: precision must be int, got {type(precision).__name__}")

src/confluent_kafka/schema_registry/rules/cel/timestamp_funcs.py:132

• TimestampType is a datetime subclass, so this early return bypasses the naive-datetime rejection below. Avro local-timestamp-* values are wrapped as TimestampType by _value_to_cel, meaning timestamp(message.local) accepts a zone-less value despite the stated limitation; formatting can then interpret it using the host timezone. Reject naive TimestampType values as well (and prevent the boundary from wrapping local timestamps as ordinary CEL timestamps).

    if isinstance(v, celtypes.TimestampType):
        return v
    if isinstance(v, Datetime):
        if v.tzinfo is None:
            # Avro local-timestamp-* logical types produce naive datetimes that

• Files reviewed: 31/34 changed files
• Comments generated: 3
• Review effort level: Balanced

[Copilot]

🟡 Changes recommended

Variant subviews can serialize incorrectly, and protobuf compatibility and JSON-name handling have unresolved defects.

Once you've addressed the issues Copilot identified, you can request another Copilot review.

Review details

Files not reviewed (2)

• src/confluent_kafka/schema_registry/confluent/type/decimal_pb2.py: Generated file
• src/confluent_kafka/schema_registry/confluent/type/variant_pb2.py: Generated file

Suppressed comments (1)

src/confluent_kafka/schema_registry/rules/cel/protobuf_result_writer.py:149

• fields_by_camelcase_name covers protobuf's derived lower-camel spelling, but not an explicitly configured json_name. A transform map using such a field's actual JSON name is silently treated as unknown and dropped, contrary to this method's contract. Resolve against each field's json_name property.

    fd = desc.fields_by_name.get(name)
    if fd is not None:
        return fd
    return desc.fields_by_camelcase_name.get(name)

• Files reviewed: 36/39 changed files
• Comments generated: 6
• Review effort level: Balanced

[Copilot]

🔵 Needs a closer look

Variant write-back, naive timestamp handling, legacy import compatibility, and the stated field-level Variant contract have unresolved issues.

Review details

Files not reviewed (2)

• src/confluent_kafka/schema_registry/confluent/type/decimal_pb2.py: Generated file
• src/confluent_kafka/schema_registry/confluent/type/variant_pb2.py: Generated file

Suppressed comments (4)

src/confluent_kafka/schema_registry/rules/cel/protobuf_result_writer.py:239

• A navigated Variant shares its parent's value buffer and starts at value.pos, so copying value.value here serializes the parent root instead of the selected subtree. This message-level write-back path needs the same standalone_value_bytes() handling already used by the Avro and common protobuf encoders.

        target.value = bytes(value.value)

src/confluent_kafka/schema_registry/rules/cel/timestamp_funcs.py:129

• This early return accepts a timezone-naive TimestampType before the following datetime guard can reject it. Avro local-timestamp-* values are converted to TimestampType at the existing boundary, so timestamp(message.local) bypasses the documented no-zone rejection and can later be interpreted using the host timezone. Check tzinfo here as well.

    if isinstance(v, celtypes.TimestampType):
        return v

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

• The canonical import replaces the previous confluent_kafka.schema_registry.confluent.types.decimal_pb2 module, which is deleted in this PR. Existing generated protobuf code and applications importing that shipped path will now fail at import time; dependency-name rewriting only helps registry descriptors. Keep the old Python module as a re-export compatibility shim to the canonical module.

from confluent_kafka.schema_registry.confluent.type import decimal_pb2, variant_pb2

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

• This deliberately makes protobuf Variant fields unreachable to CEL_FIELD rules (and the new test asserts that skip), but the PR description promises Variant marshalling for both CEL_FIELD and message-level CEL across Protobuf and Avro. Either implement the advertised field-level Variant boundary or narrow the documented contract so users are not told this rule mode works.

# Variant is deliberately *not* a leaf: it is a record in Avro too, so skipping it is the
# behaviour that matches, and a variant is reached with a message-level CEL rule instead.

• Files reviewed: 36/39 changed files
• Comments generated: 0 new
• Review effort level: Balanced

[Copilot]

🟡 Changes recommended

Variant subtree serialization can retain excluded sibling data, alongside timestamp validation and compatibility issues.

Once you've addressed the issues Copilot identified, you can request another Copilot review.

Review details

Files not reviewed (2)

• src/confluent_kafka/schema_registry/confluent/type/decimal_pb2.py: Generated file
• src/confluent_kafka/schema_registry/confluent/type/variant_pb2.py: Generated file

Suppressed comments (5)

Previously missed (2) — in code that hasn't changed since the last review.

src/confluent_kafka/schema_registry/rules/cel/timestamp_funcs.py:116

• This accepts malformed protobuf timestamps by normalizing an invalid nanos value (for example, seconds=0, nanos=-1 becomes a valid pre-epoch datetime), and out-of-range seconds escape as raw OverflowError. Protobuf Timestamp requires 0 <= nanos < 1_000_000_000; validate that invariant and normalize datetime range failures to CELEvalError, as the epoch overload does.
  src/confluent_kafka/schema_registry/rules/cel/variant_funcs.py:418
• The registered function set leaves Variant equality to Variant.__eq__, which compares encoded bytes and makes two separately parsed identical Variants equal. That contradicts the PR contract that == is identity (and that the JVM exposes no Variant equality operator). Either implement the documented identity/no-overload behavior or update the stated cross-client semantics; the current tests explicitly pin encoding equality instead.

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

• standalone_value_bytes() does not isolate a navigated node: it slices from pos to the end of the shared parent buffer, so a selected field can still serialize all later sibling payloads (and its metadata still contains the parent's key dictionary). For example, selecting a from {"a":1,"secret":"TOPSECRET"} produces a value that decodes as 1 but can still carry TOPSECRET on the wire. Re-encode/compact the selected subtree, including its metadata, before Avro write-back.

        # standalone_value_bytes, not .value: a navigated sub-variant's own value starts at
        # its position, and .value is the whole shared buffer.
        return {"metadata": data.metadata, "value": data.standalone_value_bytes()}

src/confluent_kafka/schema_registry/rules/cel/protobuf_result_writer.py:149

• fields_by_camelcase_name does not represent an explicitly configured protobuf json_name; it only covers the derived camel-case name. A transform result using a custom JSON name is therefore silently treated as an unknown field and dropped, despite this writer promising protobuf JSON-name resolution. Match FieldDescriptor.json_name instead.

    fd = desc.fields_by_name.get(name)
    if fd is not None:
        return fd
    return desc.fields_by_camelcase_name.get(name)

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

• Moving the generated module while deleting confluent.types.decimal_pb2 breaks existing application code and previously generated protobuf modules that import the old Python path; dependency rewriting only helps dynamically loaded registry descriptors. Keep a deprecated compatibility module at the old path that re-exports the canonical descriptor/message.

from confluent_kafka.schema_registry.confluent.type import decimal_pb2, variant_pb2

• Files reviewed: 36/39 changed files
• Comments generated: 2
• Review effort level: Balanced

[Copilot]

🔵 Needs a closer look

It breaks the legacy Decimal Python import path and leaks IndexError for malformed Decimal Variant payloads.

Review details

Files not reviewed (2)

• src/confluent_kafka/schema_registry/confluent/type/decimal_pb2.py: Generated file
• src/confluent_kafka/schema_registry/confluent/type/variant_pb2.py: Generated file

Suppressed comments (2)

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

• Moving the generated class without retaining the old Python module breaks existing callers that import confluent_kafka.schema_registry.confluent.types.decimal_pb2; the dependency-name rewrite only preserves serialized schema imports and cannot prevent that ModuleNotFoundError. Please keep a compatibility module at the old package path that re-exports Decimal/DESCRIPTOR from the canonical module (and deprecate it if desired).

from confluent_kafka.schema_registry.confluent.type import decimal_pb2, variant_pb2

src/confluent_kafka/schema_registry/confluent/type/variant_utils.py:507

• A malformed decimal Variant containing only its header reaches this direct indexing operation and raises IndexError, even though malformed Variant payloads are documented to raise VariantError. Validate the scale byte before reading it so malformed wire data follows the codec's error contract.

    def get_decimal(self) -> decimal.Decimal:
        _, type_info = self._primitive_info()
        scale = self.value[self.pos + 1]

• Files reviewed: 36/39 changed files
• Comments generated: 0 new
• Review effort level: Balanced

[sonarqube-confluent]

Quality Gate passed

Issues
44 New issues
1 Fixed issue
0 Accepted issues

Measures
0 Security Hotspots
81.9% Coverage on New Code
0.0% Duplication on New Code

See analysis details on SonarQube

[Copilot]

🔵 Needs a closer look

Four unresolved correctness and compatibility issues can cause import failures, silent field loss, or invalid value handling.

Review details

Files not reviewed (2)

• src/confluent_kafka/schema_registry/confluent/type/decimal_pb2.py: Generated file
• src/confluent_kafka/schema_registry/confluent/type/variant_pb2.py: Generated file

Suppressed comments (4)

Previously missed (1) — in code that hasn't changed since the last review.

src/confluent_kafka/schema_registry/rules/cel/timestamp_funcs.py:116

• Protobuf permits constructing a Timestamp object with invalid wire values, but this path neither validates the documented seconds/nanos ranges nor normalizes failures as CEL errors. For example, nanos=-1 is silently normalized to the prior microsecond, while an out-of-range seconds value leaks OverflowError. Reject values outside the protobuf Timestamp contract before constructing the datetime.

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

• Switching to the canonical module removes the previously shipped confluent_kafka.schema_registry.confluent.types.decimal_pb2 import path. Existing applications that import that generated message directly will now fail at import time even though schema dependency names are rewritten. Keep a compatibility module at the old path that re-exports Decimal and DESCRIPTOR while using the canonical descriptor internally.

from confluent_kafka.schema_registry.confluent.type import decimal_pb2, variant_pb2

src/confluent_kafka/schema_registry/rules/cel/protobuf_result_writer.py:149

• fields_by_camelcase_name only covers the derived lower-camel spelling, not an explicitly configured protobuf json_name. A transform returning that valid JSON field name is therefore treated as unknown and silently drops the field, contrary to this writer's documented declared-name-or-JSON-name behavior. Resolve against each field's json_name property instead.

    return desc.fields_by_camelcase_name.get(name)

src/confluent_kafka/schema_registry/rules/cel/variant_funcs.py:110

• An absent protobuf/Avro Variant has both buffers empty, but this treats any empty metadata as absence even when value contains data. That silently converts a malformed or partially populated Variant to CEL null, allowing null-guarded rules to pass instead of reporting corrupt input. Only return null when both buffers are empty; reject a non-empty value without metadata.

    if not metadata:
        return None
    return Variant(value, metadata)

• Files reviewed: 36/39 changed files
• Comments generated: 0 new
• Review effort level: Balanced