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. A Variant is
converted at the boundary too, but only a message-level rule reaches it.

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]

🔵 Needs a closer look

Variant write-back can retain discarded sensitive bytes, and empty Avro variants and wide-scale zero decimals are handled incorrectly.

Review details

Files not reviewed (3)

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

Suppressed comments (3)

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

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

• Zero is compact at every scale, but this guard treats a positive scale as if it appended digits. For example, decimal_to_protobuf(Decimal("0"), 1_000_000_000) is rejected even though the equivalent negative-scale cases and BigDecimal.setScale accept it immediately. Skip both the width check and power computation when the coefficient is zero.

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

• The logical reader constructs Variant unconditionally, whose constructor reads metadata[0]. Consequently an Avro variant record with both byte fields empty fails during deserialization instead of reaching CEL as the documented absent/null value; the existing absent-Avro test bypasses this reader by passing a dict directly. Handle the all-empty wire representation here, while rejecting a non-empty value with missing metadata.

def _variant_from_avro(data, writer_schema, reader_schema=None):  # noqa: ARG001
    return Variant(bytes(data["value"]), bytes(data["metadata"]))

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

• This slice includes every byte after the selected node, including later sibling values. Thus writing back a navigated child may decode as the child but still transmit discarded sensitive data (the test's TOPSECRET remains in the serialized payload). Produce the exact self-delimiting node encoding—and compact/remap metadata where needed—before using it at the Avro and Protobuf write-back boundaries.

        Like all of those, this slices to the end of the buffer rather than to the node's exact
        extent, so a navigated value still carries its later siblings' bytes. Decoding ignores
        them - the encoding is self-delimiting.
        """
        return self.value[self.pos:] if self.pos else self.value

• Files reviewed: 40/43 changed files
• Comments generated: 0 new
• Review effort level: Balanced

[Copilot]

🔵 Needs a closer look

Variant write-back can retain discarded sibling data, and Decimal double conversion does not preserve JVM scale semantics.

Review details

Files not reviewed (3)

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

Suppressed comments (3)

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

• Absence is inferred from metadata alone, so a corrupt decoded record such as {metadata: b"", value: b"..."} is silently converted to CEL null and its payload is ignored. Only the all-empty protobuf/Avro default represents absence; one empty component should be reported as malformed input rather than passing null-aware rules.

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

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

• Using Python str(float) does not preserve the JVM BigDecimal.valueOf(double) encoding even when the numeric value matches. For example, Java formats 1e7 as 1.0E7 (scale -6), while Python formats it as 10000000.0 (scale 1); subsequent protobuf write-back therefore emits different unscaled bytes/scale, despite scale being part of this API's value contract. Convert doubles through a Java-compatible Double.toString representation and add scale assertions around the notation thresholds.

    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/confluent/type/variant_utils.py:369

• This slice is not actually standalone: for a navigated child it includes every byte after that child, including later siblings. Writing variants.field(doc, "a") from {"a":1,"secret":"TOPSECRET"} therefore still puts TOPSECRET on the wire even though decoding shows only 1, defeating the safe-subtree use case described by the new tests. Compute the selected node's encoded extent (and ideally compact/remap its metadata) before serialization, and assert the discarded sibling bytes are absent.

        Like all of those, this slices to the end of the buffer rather than to the node's exact
        extent, so a navigated value still carries its later siblings' bytes. Decoding ignores
        them - the encoding is self-delimiting.
        """
        return self.value[self.pos :] if self.pos else self.value

• Files reviewed: 40/43 changed files
• Comments generated: 0 new
• Review effort level: Balanced

[Copilot]

🟡 Changes recommended

Variant write-back can retain removed sibling data, alongside timestamp and decimal edge-case defects.

Get a fresh assessment by requesting another Copilot review.

Review details

Files not reviewed (3)

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

Suppressed comments (5)

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

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

• A zero coefficient does not need a BigInteger expansion, so this guard incorrectly rejects valid positive scales such as decimal_to_protobuf(Decimal("0"), 1_000_000_000). That scale fits the wire int32, and BigDecimal.ZERO.setScale(...) succeeds immediately; skip both the width guard and power construction for zero, as the narrowing branch already does.

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

• This writes standalone_value_bytes(), which is only a suffix from pos and therefore contains all later sibling bytes. A CEL transform selecting a safe child can still place omitted secret data and the parent's metadata on the Protobuf wire even though decoding shows only the child. The standalone representation needs an exact node encoding and only its required metadata.

    result.value = value.standalone_value_bytes()

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

• A navigated Variant's standalone_value_bytes() is a suffix of the parent buffer, not the selected node's exact bytes, so this message-level writer can retain later sibling payloads that the CEL result removed. This is observable data exposure on the serialized wire; rebuild an exact standalone Variant (including minimal/remapped metadata) before assignment.

        target.value = bytes(value.standalone_value_bytes())

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

• The PR description promises schema-to-CEL marshalling for both CEL_FIELD and message-level CEL, including Variant, but this explicitly excludes Variant from protobuf field rules; the added field test likewise asserts that a tagged Variant is skipped. Consequently variants.* cannot be used on a Variant in CEL_FIELD. Either treat Variant as a leaf and rebuild it on write-back, or narrow the stated API scope/compatibility claim.

# 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.

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

• Avro values pass through _value_to_cel first, which wraps every datetime—including a naive local-timestamp-*—as TimestampType. This early return therefore bypasses the tzinfo is None rejection below, so timestamp(localValue) accepts the exact zoneless shape the PR says must be refused. Validate tzinfo before returning an existing TimestampType.

    if isinstance(v, celtypes.TimestampType):

• Files reviewed: 40/43 changed files
• Comments generated: 1
• Review effort level: Balanced

[Copilot]

🔵 Needs a closer look

Variant write-back can retain removed sibling data, and Avro local timestamps bypass the documented rejection path.

Review details

Files not reviewed (3)

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

Suppressed comments (2)

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

• standalone_value_bytes() returns the entire suffix from the selected node, so writing a navigated child also writes every later sibling's raw bytes. For example, selecting a from {"a":1,"secret":"TOPSECRET"} still places TOPSECRET in the serialized protobuf value; decoding displays only 1, but the supposedly removed data remains on the wire. Please serialize only the selected node's encoded extent (and use that exact slice in all Avro/protobuf write-back paths).

    # standalone_value_bytes, not .value: a navigated sub-variant's own value starts at its
    # position, and .value is the whole shared buffer.
    result.value = value.standalone_value_bytes()

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

• The naive-datetime rejection here is bypassed for actual Avro local-timestamp-* fields: msg_to_cel() recursively calls cel_executor._value_to_cel, which constructs TimestampType(msg) for every datetime before timestamp(...) is invoked. Such fields therefore still become instants (using host-local timezone behavior) even though the PR declares them unsupported. Reject naive datetimes at the schema-to-CEL boundary as well, or preserve them as a non-timestamp value.

    if isinstance(v, Datetime):
        if v.tzinfo is None:
            # Avro local-timestamp-* logical types produce naive datetimes that
            # carry no timezone — refuse rather than silently picking UTC.
            raise celpy.CELEvalError(
                "timestamp: naive datetime (no timezone) cannot be converted. "
                "Use the regular timestamp-* logical type (UTC by spec), or pass "
                "an offset-adjusted epoch value via timestamp(value, precision)."
            )

• Files reviewed: 40/43 changed files
• Comments generated: 0 new
• Review effort level: Balanced

[Copilot]

🔵 Needs a closer look

Avro local timestamps currently bypass the intended naive-datetime rejection and enter CEL as timestamps.

Review details

Files not reviewed (3)

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

Suppressed comments (1)

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

• The naive-datetime rejection is bypassed for Avro local-timestamp-* values. msg_to_cel() first routes every datetime through _value_to_cel (cel_executor.py:205-207), which wraps it as TimestampType; this function then returns at this branch before checking tzinfo. Consequently a local timestamp is exposed to CEL despite the PR's stated limitation. Preserve raw naive datetimes until this dispatch, or reject them at the boundary before constructing TimestampType.

    if isinstance(v, celtypes.TimestampType):
        return v
    if isinstance(v, Datetime):
        if v.tzinfo is None:

• Files reviewed: 40/43 changed files
• Comments generated: 0 new
• Review effort level: Balanced

[Copilot]

🟡 Changes recommended

Decimal conversion, custom Protobuf JSON names, and oversized Variant decimals have unresolved correctness issues.

Get a fresh assessment by requesting another Copilot review.

Review details

Files not reviewed (3)

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

Suppressed comments (2)

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

• decimal(double) does not preserve the JVM BigDecimal.valueOf encoding because Python's str(float) is not Java's Double.toString. For example, Java converts 1e-7 through "1.0E-7" (scale 8, rendering 0.00000010), while this code converts Python's "1e-07" (scale 7, rendering 0.0000001); 10000000.0 similarly renders differently. Since scale is part of the Decimal contract, use a Java-compatible double-to-string conversion before constructing the Decimal.

        # 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/protobuf_result_writer.py:146

• This resolves only the computed camelCase name, not an explicitly configured protobuf json_name. A transform map using a custom JSON name is therefore treated as an unknown field and silently dropped, despite the documented JSON-name support. Resolve against each field's json_name property instead.

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

• Files reviewed: 40/43 changed files
• Comments generated: 1
• Review effort level: Balanced

[Copilot]

🔵 Needs a closer look

Explicitly configured protobuf JSON field names can be silently dropped during CEL transform write-back.

Review details

Files not reviewed (3)

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

Suppressed comments (1)

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

• This lookup does not handle explicitly overridden protobuf JSON names. fields_by_camelcase_name is keyed by each field's derived camel-case name, whereas FieldDescriptor.json_name may be set to an unrelated value (for example, foo_bar [json_name = "amount"]). A CEL transform returning that valid JSON key is therefore treated as an unknown field and silently dropped, unlike the JVM JSON write-back. Resolve against field.json_name instead.

    return desc.fields_by_camelcase_name.get(name)

• Files reviewed: 40/43 changed files
• Comments generated: 0 new
• Review effort level: Balanced

[sonarqube-confluent]

Quality Gate passed

Issues
49 New issues
1 Fixed issue
0 Accepted issues

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

See analysis details on SonarQube