confluent: fix schema_registry_encode rejecting decoder output - #4704
confluent: fix schema_registry_encode rejecting decoder output#4704twmb wants to merge 5 commits into
Conversation
schema_registry_decode emits Avro JSON, where a bytes value is one codepoint per byte: the unscaled decimal 0x21 comes out as "!". The encoder handed that string straight to the native Encode, which reads a Go string as UTF-8 and accepts only numeric text for a decimal, so under twmb/avro v1.8.0 it fails with `invalid decimal string "!"`. Every decimal backed by bytes or fixed was unencodable, in both avro_raw_json modes. The encoder now parses the message with DecodeJSON and encodes the native value it returns. DecodeJSON takes tagged and bare unions alike, so one shared path still serves both modes. What it rejects still falls through to Encode, the more permissive reader: RFC 3339 timestamp strings and time.Time have no Avro JSON spelling and CDC sources send them. Also bumps twmb/avro to v1.8.0. A new test decodes Avro binary and re-encodes it to the same bytes in both modes; the logical-type encoder tests now assert exact bytes instead of just a wire header.
![claude[bot]](https://avatars.githubusercontent.com/in/1236702?s=60&v=4){kind=small}
| return func(m *service.Message) error { | ||
| data, err := m.AsStructuredMut() | ||
| if err != nil { | ||
| return fmt.Errorf("extracting structured data: %w", err) | ||
| var native any | ||
| b, err := m.AsBytes() | ||
| if err != nil || schema.DecodeJSON(b, &native) != nil { | ||
| if native, err = m.AsStructuredMut(); err != nil { | ||
| return fmt.Errorf("extracting structured data: %w", err) | ||
| } | ||
| } |
There was a problem hiding this comment.
Routing through m.AsBytes() before DecodeJSON changes behaviour for messages that are structured rather than raw bytes, and it can corrupt bytes/fixed fields silently.
Previously the closure went straight to AsStructuredMut(), so a []byte value nested in the structure was handed to schema.Encode intact. Now AsBytes() runs first, which JSON-serialises the structured value — encoding/json renders a nested []byte as base64. DecodeJSON then reads that base64 text with Avro JSON's one-codepoint-per-byte semantics, succeeds (any string is a valid Avro JSON bytes value), and never falls through to the AsStructuredMut path. The encoded field is the base64 characters, not the original bytes, with no error surfaced.
Concrete case: a mapping such as root.data = this.b64.decode("base64") feeding a schema whose data field is bytes — decode yields []byte in the structured value. Same applies downstream of schema_registry_decode with preserve_logical_types: true, which sets structured data containing []byte for plain bytes fields (a record with no timestamp field won't hit the Encode fallback that would otherwise mask this).
Suggested fix: only take the DecodeJSON path when the message payload is genuinely a raw byte payload (Avro JSON text), and keep structured messages on the AsStructuredMut + Encode path — or try AsStructuredMut first and fall back to DecodeJSON on the serialised form. Worth a test covering a plain bytes field (non-decimal) fed from a structured message, which the new round-trip test doesn't reach since testSchemaLogicalTypes only exercises bytes.decimal from a bytes-payload message.
Refs internal/impl/confluent/serde_avro.go#L194-L201 and the decoder's structured path at serde_avro.go#L277-L279.
There was a problem hiding this comment.
Confirmed and fixed in b672342.
service.Message answers this directly: HasStructured() / HasBytes() report which forms are cached without converting either way. The encoder now reads a message as Avro JSON only when !m.HasStructured(); everything else goes to AsStructuredMut + Encode. No serialise-and-sniff, so the base64 round trip never happens.
Trade-off, since neither reader is a superset of the other: a message holding both forms — a raw payload something upstream read via AsStructured(), which caches the parse — takes Encode. That is the reader it had before this PR, so nothing regresses there, and a bytes-backed decimal on that path fails loudly (invalid decimal string) rather than encoding wrong bytes. I picked that over preferring the cached raw bytes because the failure modes are not symmetric: preferring bytes reintroduces exactly the silent corruption you found, preferring structured only loses the fix in a corner and loses it noisily. I also skipped an "Encode failed, retry DecodeJSON on the cached bytes" fallback for the same reason — it would resurrect the base64 path on the error branch.
Tests, in processor_schema_registry_encode_test.go:
TestSchemaRegistryEncodeAvroMessageFormruns both forms of the same value against the same expected wire bytes: a structured[]byteand Avro JSON codepoints for a plainbytesfield, the same forbytes+fixed+ decimals backed by each, and timestamps astime.Time, as RFC 3339 text (structured and raw), and as Avro JSON micros.- The non-ASCII case is explicit:
rawise9 00 7f 21, spelled in Avro JSON as the codepoints U+00E9, U+0000, U+007F,!.Encodewrites U+00E9 as its two UTF-8 bytesc3 a9. TestSchemaRegistryAvroPreserveLogicalTypesRoundTripis the pipeline you named — decode withpreserve_logical_typesfeeding the encoder — over both a bytes-only schema and the bytes+fixed one.
Both breaks were confirmed failing before I trusted the tests. With DecodeJSON first, expected \x00\x00\x00\x00\x04\b\xe9\x00\x7f! vs actual \x00\x00\x00\x00\x04\x106QB/IQ== — your base64 corruption, on the plain bytes field and on the preserve_logical_types round trip. With AsStructuredMut first, actual \x00\x00\x00\x00\x04\n<c3><a9>\x00\x7f! — U+00E9 as two bytes, length prefix 5 instead of 4 — plus the decimal failures across both avro_raw_json modes.
One note on "worth a test covering a plain bytes field": it has to be a schema with no fixed field to reproduce. base64 of 4 bytes is 8 characters, so a fixed(4) size check rejects the serialised form and the message falls through to Encode by accident, masking the bug. Hence the separate single-field testSchemaRawBytes.
avro v1.8.0 changed Schema.Root() to return *avro.SchemaNode, and the pinned iceberg-go predates apache/iceberg-go#1843, which adapts to it. It compiles against the version it pins, but Go builds one version per module graph, so raising avro here broke iceberg-go's own sources and typechecking cascaded into internal/impl/iceberg/icebergx. The bumped transitive requirements come from iceberg-go's go.mod and are identical whether it is pinned at the merge commit or at main.
|
Pushed a second commit bumping CI failed on TYPECHECK rather than a lint rule: avro v1.8.0 changed The go.mod churn is larger than the one line because iceberg-go's own go.mod requires newer otel, grpc, protobuf and cloud SDKs. That set is identical whether iceberg-go is pinned at the merge commit or at main, so I took main.
|
| require.NoError(t, err) | ||
| defer func() { _ = encoder.Close(t.Context()) }() | ||
|
|
||
| for _, wire := range []string{ |
There was a problem hiding this comment.
Test coverage gap: the fixed-backed decimal half of the fix is never exercised.
The commit message and the new code comment both scope the bug to "every decimal backed by bytes or fixed" (serde_avro.go#L184-L188), but testSchemaLogicalTypes — the only schema this round-trip test uses — contains just a bytes decimal (processor_schema_registry_decode_test.go#L195-L207). No test in internal/impl/confluent puts a fixed type through the encode path: the three "type": "fixed" occurrences in the package are in avro_walker_test.go and ecs_avro_test.go, which cover schema walking / common-schema conversion, not newAvroEncoder.
fixed differs from bytes on the wire (no length prefix, size-checked), so it is a distinct path through DecodeJSON → Encode and can regress independently.
Suggested fix: add a fixed-backed decimal field (e.g. {"type":"fixed","name":"Dec","size":16,"logicalType":"decimal","precision":38,"scale":8}) to a schema used by this round-trip test, so both branches named in the fix are pinned.
Per CONTRIBUTING.md §1.3.2 — "Tests should cover end-to-end functionality and prove that the connector works across supported configurations."
There was a problem hiding this comment.
Right, fixed was unexercised. Added in b672342.
Rather than change testSchemaLogicalTypes — its expected wire bytes are asserted verbatim by three other tests — the round trip is now table-driven over schemas, and the second one is testSchemaBytesFixed:
{"name": "raw", "type": "bytes"}
{"name": "raw_fixed", "type": {"type": "fixed", "name": "Raw4", "size": 4}}
{"name": "dec_bytes", "type": {"type": "bytes", "logicalType": "decimal", "precision": 16, "scale": 2}}
{"name": "dec_fixed", "type": {"type": "fixed", "name": "Dec", "size": 16, "logicalType": "decimal", "precision": 38, "scale": 8}}That covers both halves of the fix and both wire shapes side by side: dec_bytes carries a length prefix, dec_fixed is 16 raw bytes with the unscaled 0xbc614e left-padded and no prefix. Plain raw/raw_fixed ride along so the codepoint spelling is pinned without a logical type on top, and raw is e9 00 7f 21 so a byte above 0x7f is in there.
Same schema runs through TestSchemaRegistryEncodeAvroMessageForm and TestSchemaRegistryAvroPreserveLogicalTypesRoundTrip, so fixed is pinned through the structured Encode path as well as the DecodeJSON one.
Verified failing first: with the encoder reverted to AsStructuredMut + Encode, both avro_raw_json modes of the bytes_and_fixed round trip fail.
v0.6 honours the "endpoints" field of the REST /v1/config response. The spec makes the field optional and reads its absence as "the default set only", which excludes the two HEAD exists endpoints, so a client talking to a server that omits it now answers an existence check with GET (load) instead of HEAD. catalogx's mock server omitted the field, so CheckTableExists and CheckNamespaceExists stopped issuing the HEAD their tests count. The production wrappers need no change — they retry on an auth error whatever verb the library picks — so the mock now advertises its endpoints, restoring the HEAD path those tests were written for, and a new test covers the GET fallback that a non-advertising catalog now takes. Also silences the SA1019 on Transform.ToHumanStr rather than moving to ToHumanStrType. The two differ only for identity on timestamptz, where ToHumanStrType appends "+00:00" and would thereby rename the partition directory of every identity-partitioned timestamptz column, splitting new writes from existing files and diverging from the C++ datalake writer this file mirrors. That is a storage-layout change, not a dependency bump.
|
Pushed The three catalogx failures: deliberate upstream behavior change, spec-correct. v0.6 honours the Our mock server omitted The SA1019: not adopted, deliberately. Scoped verification, all passing: |
The encoder had two readers and picked between them by serialising the
message and seeing whether DecodeJSON accepted the result. For a
structured message that is silently wrong: AsBytes runs encoding/json,
which writes a nested []byte as base64 text, and DecodeJSON accepts any
string as a bytes value, so a `bytes` field arrived on the wire as its
base64 characters with no error. A mapping like
`root.data = this.b64.decode("base64")` hits it, as does anything
downstream of schema_registry_decode with preserve_logical_types.
HasStructured answers the question the serialisation was standing in
for, and answers it without converting either form, so only a message
that really is a raw payload is read as Avro JSON. Structured messages
go to Encode, the only reader that takes the Go values they carry:
[]byte for bytes and fixed, time.Time or RFC 3339 text for timestamps.
A message holding both forms goes to Encode, the reader it had before
either was a choice.
Tests pin both readers against the same wire bytes: a structured []byte
and Avro JSON codepoints (including 0xe9, which Encode would spell as
two UTF-8 bytes) for the same bytes field, timestamps as time.Time, RFC
3339 text and micros, and a preserve_logical_types round trip. The
decode/encode round trip gains a schema with plain bytes, plain fixed,
and decimals backed by each, so the fixed half of the original fix is
exercised too.
Picking the reader by message form serves the shapes a message is born with, but form is a hint, not a fact. Reading a message structurally drops its raw payload, so schema_registry_decode followed by any mapping hands the encoder Avro JSON values in a structured message, which the form check reads the plain way - and the bytes-backed decimal fails exactly as it did before either reader was a choice. Nothing on the message can settle it: after the parse, a decoder payload and a mapping-built tree are indistinguishable, and the string "!" is a valid value under both readers with two different meanings. So the choice becomes declarable: avro.input_encoding. auto stays the default and its path is unchanged; avro_json reads every message as Avro JSON; native reads every message as Go and plain JSON values. avro_json never trusts the message byte cache. AsBytes serialises the structured form the moment anything asks for the payload - the subject interpolation of this very processor included - so a cached payload can be the base64 spelling of the very []byte that must not be read as codepoints. The bytes to decode therefore come from marshalling the tree itself, after checking that every value in it is one encoding/json spells unambiguously. A tree carrying []byte or time.Time is refused with the mode that can encode it named in the error, a value JSON cannot spell at all (NaN, the infinities) fails naming the serialisation as the cause, and in both cases the message reaches the error path with its contents intact. Tests pin the reading of each mode, the modes agreeing on the wire bytes wherever they should, the cached-bytes refusal, the config plumbing with its default, and that an unrecognised mode reads as auto.
2 participants
schema_registry_decode emits Avro JSON, where a bytes value is one
codepoint per byte: the unscaled decimal 0x21 comes out as "!". The
encoder handed that string straight to the native Encode, which reads a
Go string as UTF-8 and accepts only numeric text for a decimal, so under
twmb/avro v1.8.0 it fails with
invalid decimal string "!". Everydecimal backed by bytes or fixed was unencodable, in both avro_raw_json
modes.
The encoder now picks its reader by the form each message arrives in. A
raw payload is parsed with DecodeJSON, which implements Avro JSON's bytes
and fixed semantics and takes tagged and bare unions alike, so one path
serves both modes; raw JSON it rejects still falls through to Encode. A
structured message goes to Encode directly, the only reader for the Go
values it carries — []byte for bytes and fixed, time.Time and RFC 3339
text for timestamps, a shape CDC sources emit and Avro JSON cannot spell.
Form is a hint rather than a fact, though: reading a message structurally
drops its raw payload, so decode → mapping → encode reaches the encoder
with Avro JSON values in structured form, and nothing on the message can
tell that tree from one a mapping built — the string "!" is a valid value
under both readers with two different meanings. For pipelines that know
their input, a new
avro.input_encodingfield declares it:auto(thedefault) keeps the form-based choice,
avro_jsonreads every message asAvro JSON,
nativereads every message as Go and plain JSON values.avro_jsonmarshals the structured tree itself rather than trusting themessage's byte cache — AsBytes spells a nested []byte as base64 text,
which DecodeJSON would happily read as codepoints — and refuses values
Avro JSON cannot spell, naming a mode that can in the error.
Also bumps twmb/avro to v1.8.0, dragging an iceberg-go bump and its v0.6
REST endpoint negotiation along. New tests decode Avro binary and
re-encode it to the same bytes in both modes, directly and through an
intervening parse; pin which reader each form and mode gets, with the
modes agreeing on the wire bytes wherever they should; and assert exact
bytes throughout, fixed-backed decimals included.
Two silent changes worth knowing about under the default
automode,both consequences of reading a raw payload as Avro JSON. A non-ASCII
string in a
bytesfield now encodes as one byte per codepoint ratherthan as UTF-8, so
"é"goes from04 c3 a9to02 e9; both pathssucceed, and the new encoding is the Avro JSON semantics. A decimal
supplied as numeric text is now read as codepoint bytes rather than as
decimal notation: when those bytes overrun the declared precision it
fails, and when they fit it silently encodes the wrong number —
{"d":"3.33"}against precision 16, scale 2 becomes 8588628.99. Thedecoder never emits that form, so round trips are unaffected;
hand-authored input that means decimal notation should set
avro.input_encoding: native.