Skip to content

confluent: fix schema_registry_encode rejecting decoder output #4704

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
twmb wants to merge 5 commits into
base: main
Choose a base branch
from
Open

confluent: fix schema_registry_encode rejecting decoder output #4704

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
1c1d1a7
confluent: fix schema_registry_encode rejecting decoder output
twmb Aug 17, 2026
5bbed70
deps: bump iceberg-go for avro v1.8.0
twmb Aug 17, 2026
9881d90
iceberg: adapt to iceberg-go v0.6 REST endpoint negotiation
twmb Aug 18, 2026
b672342
confluent: pick the Avro encode reader by message form
twmb Aug 18, 2026
12a2cff
confluent: let the Avro encode reader be declared, not just inferred
Jeffail Aug 19, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion go.mod
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -170,7 +170,7 @@ require (
github.com/timeplus-io/proton-go-driver/v2 v2.1.4
github.com/tmc/langchaingo v0.1.14
github.com/trinodb/trino-go-client v0.333.0
github.com/twmb/avro v1.7.3-0.20260513193503-1e5c2a3fc070
github.com/twmb/avro v1.8.0
github.com/twmb/franz-go v1.20.7
github.com/twmb/franz-go/pkg/kadm v1.17.2
github.com/twmb/franz-go/pkg/kmsg v1.12.0
Expand Down
4 changes: 2 additions & 2 deletions go.sum
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1742,8 +1742,8 @@ github.com/trivago/grok v1.0.0/go.mod h1:9t59xLInhrncYq9a3J7488NgiBZi5y5yC7bss+w
github.com/trivago/tgo v1.0.7 h1:uaWH/XIy9aWYWpjm2CU3RpcqZXmX2ysQ9/Go+d9gyrM=
github.com/trivago/tgo v1.0.7/go.mod h1:w4dpD+3tzNIIiIfkWWa85w5/B77tlvdZckQ+6PkFnhc=
github.com/tv42/httpunix v0.0.0-20150427012821-b75d8614f926/go.mod h1:9ESjWnEqriFuLhtthL60Sar/7RFoluCcXsuvEwTV5KM=
github.com/twmb/avro v1.7.3-0.20260513193503-1e5c2a3fc070 h1:gYa95NoqeXYOMIVIe2/YcbC/l0s5q+wQ70Zo+TYQU4k=
github.com/twmb/avro v1.7.3-0.20260513193503-1e5c2a3fc070/go.mod h1:X0fT1dY2xcbV4YuCE4mYro+qljHl4kUF5uA/2z1rgSk=
github.com/twmb/avro v1.8.0 h1:UMWLg+nH4P3yad5Om7yFSohYLy2RG1s7BcFFiOvmK9Q=
github.com/twmb/avro v1.8.0/go.mod h1:X0fT1dY2xcbV4YuCE4mYro+qljHl4kUF5uA/2z1rgSk=
github.com/twmb/franz-go v1.20.7 h1:P4MGSXJjjAPP3NRGPCks/Lrq+j+twWMVl1qYCVgNmWY=
github.com/twmb/franz-go v1.20.7/go.mod h1:0bRX9HZVaoueqFWhPZNi2ODnJL7DNa6mK0HeCrC2bNU=
github.com/twmb/franz-go/pkg/kadm v1.17.2 h1:g5f1sAxnTkYC6G96pV5u715HWhxd66hWaDZUAQ8xHY8=
Expand Down
95 changes: 81 additions & 14 deletions internal/impl/confluent/processor_schema_registry_encode_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -320,12 +320,12 @@ func TestSchemaRegistryEncodeAvroLogicalTypes(t *testing.T) {
output: "\x00\x00\x00\x00\x04\x02\x90\xaf\xce!\x02\x80\x80揪\x97\t\x02\x80\x80\xde\xf2\xdf\xff\xdf\xdc\x01\x02\x02!",
},
{
// The normalizer auto-wraps plain values for nullable unions,
// so unwrapped input that previously required lame-union format
// now succeeds. Verify via round-trip decode.
name: "message with unwrapped unions succeeds with normalizer",
// Bare union values are accepted alongside the tagged form, so
// unwrapped input that once required the tagged shape encodes to
// the same bytes.
name: "message with unwrapped unions",
input: `{"int_time_millis":35245000,"long_time_micros":20192000000000,"long_timestamp_micros":null,"pos_0_33333333":"!"}`,
output: "", // verified via round-trip below
output: "\x00\x00\x00\x00\x04\x02\x90\xaf\xce!\x02\x80\x80揪\x97\t\x00\x02\x02!",
},
{
// Wrong union key ("long.time-millis" instead of "int.time-millis")
Expand Down Expand Up @@ -356,15 +356,7 @@ func TestSchemaRegistryEncodeAvroLogicalTypes(t *testing.T) {

b, bErr := outBatches[0][0].AsBytes()
require.NoError(t, bErr)

if test.output != "" {
assert.Equal(t, test.output, string(b))
} else {
// No expected bytes — just verify valid Confluent wire
// format: magic byte + 4-byte schema ID + Avro binary.
require.Greater(t, len(b), 5, "output must have wire header")
assert.Equal(t, byte(0x00), b[0], "magic byte")
}
assert.Equal(t, test.output, string(b))
}
})
}
Expand Down Expand Up @@ -455,6 +447,81 @@ func TestSchemaRegistryEncodeAvroRawJSONLogicalTypes(t *testing.T) {
encoder.cacheMut.Unlock()
}

// TestSchemaRegistryAvroDecodeEncodeRoundTrip pins the symmetry of the two
// processors: whatever schema_registry_decode emits must re-encode, through
// schema_registry_encode on the same schema, to the exact bytes it was decoded
// from — in both avro_raw_json modes, which differ only in whether unions are
// tagged.
//
// Decimals backed by bytes are the case that broke this. Avro JSON spells a
// bytes value as one codepoint per byte, so the unscaled value 0x21 is emitted
// as "!", which the native encoder read as a decimal in decimal notation and
// rejected.
func TestSchemaRegistryAvroDecodeEncodeRoundTrip(t *testing.T) {
byID, err := json.Marshal(struct {
Schema string `json:"schema"`
}{
Schema: testSchemaLogicalTypes,
})
require.NoError(t, err)

bySubject, err := json.Marshal(struct {
Schema string `json:"schema"`
ID int `json:"id"`
}{
Schema: testSchemaLogicalTypes,
ID: 4,
})
require.NoError(t, err)

urlStr := runSchemaRegistryServer(t, func(path string) ([]byte, error) {
switch path {
case "/schemas/ids/4":
return byID, nil
case "/subjects/foo/versions/latest":
return bySubject, nil
}
return nil, errors.New("nope")
})

subj, err := service.NewInterpolatedString("foo")
require.NoError(t, err)

for _, rawJSON := range []bool{false, true} {
t.Run(fmt.Sprintf("avro_raw_json=%v", rawJSON), func(t *testing.T) {
cfg := decodingConfig{}
cfg.avro.rawUnions = rawJSON
decoder, err := newSchemaRegistryDecoder(urlStr, noopReqSign, nil, cfg, schemaStaleAfter, service.MockResources())
require.NoError(t, err)
defer func() { _ = decoder.Close(t.Context()) }()

encoder, err := newSchemaRegistryEncoder(urlStr, noopReqSign, nil, subj, rawJSON, time.Minute*10, time.Minute, service.MockResources())
require.NoError(t, err)
defer func() { _ = encoder.Close(t.Context()) }()

for _, wire := range []string{

@claude claude Bot Aug 17, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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

@twmb twmb Aug 18, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

"\x00\x00\x00\x00\x04\x02\x90\xaf\xce!\x02\x80\x80揪\x97\t\x02\x80\x80\xde\xf2\xdf\xff\xdf\xdc\x01\x02\x02!",
// Every union on its null branch.
"\x00\x00\x00\x00\x04\x00\x00\x00\x00",
} {
decoded, err := decoder.Process(t.Context(), service.NewMessage([]byte(wire)))
require.NoError(t, err)
require.Len(t, decoded, 1)

outBatches, err := encoder.ProcessBatch(t.Context(), service.MessageBatch{decoded[0]})
require.NoError(t, err)
require.Len(t, outBatches, 1)
require.Len(t, outBatches[0], 1)
require.NoError(t, outBatches[0][0].GetError())

b, err := outBatches[0][0].AsBytes()
require.NoError(t, err)
assert.Equal(t, wire, string(b))
}
})
}
}

func TestSchemaRegistryEncodeClearExpired(t *testing.T) {
urlStr := runSchemaRegistryServer(t, func(string) ([]byte, error) {
return nil, fmt.Errorf("nope")
Expand Down
27 changes: 21 additions & 6 deletions internal/impl/confluent/serde_avro.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -177,14 +177,29 @@ func (*schemaRegistryEncoder) newAvroEncoder(avroJSON string) (schemaEncoder, er
return nil, fmt.Errorf("parsing Avro schema: %w", err)
}

// Encode accepts both bare values (standard JSON) and tagged union
// maps (Avro JSON), so both avroRawJSON modes use the same path.
// Avro JSON is the canonical input: it is what schema_registry_decode
// emits (EncodeJSON) in both avroRawJSON modes — only union tagging
// differs, and DecodeJSON accepts tagged and bare unions alike — so one
// path still serves both. DecodeJSON is also the only reader that
// implements Avro JSON's bytes and fixed semantics, where a JSON string
// carries one byte per codepoint. Encode does not: it reads a Go string
// as UTF-8, mangling any byte above 0x7f, and for a decimal it accepts
// only numeric text, so every decimal backed by bytes or fixed failed to
// re-encode from what the decoder emitted.
//
// Input that is not Avro JSON falls through to Encode, which is the more
// permissive of the two: it also takes RFC 3339 strings and time.Time
// for timestamp fields, a shape CDC sources emit and Avro JSON cannot
// spell. Its error is therefore the one worth reporting.
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)
}
}
Comment on lines 243 to 251

@claude claude Bot Aug 17, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

@twmb twmb Aug 18, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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:

  • TestSchemaRegistryEncodeAvroMessageForm runs both forms of the same value against the same expected wire bytes: a structured []byte and Avro JSON codepoints for a plain bytes field, the same for bytes + fixed + decimals backed by each, and timestamps as time.Time, as RFC 3339 text (structured and raw), and as Avro JSON micros.
  • The non-ASCII case is explicit: raw is e9 00 7f 21, spelled in Avro JSON as the codepoints U+00E9, U+0000, U+007F, !. Encode writes U+00E9 as its two UTF-8 bytes c3 a9.
  • TestSchemaRegistryAvroPreserveLogicalTypesRoundTrip is the pipeline you named — decode with preserve_logical_types feeding 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.

binary, err := schema.Encode(data)
binary, err := schema.Encode(native)
if err != nil {
return err
}
Expand Down
2 changes: 1 addition & 1 deletion public/bundle/enterprise/go.mod
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -444,7 +444,7 @@ require (
github.com/substrait-io/substrait-protobuf/go v0.85.0 // indirect
github.com/tilinna/z85 v1.0.0 // indirect
github.com/timandy/routine v1.1.5 // indirect
github.com/twmb/avro v1.7.3-0.20260513193503-1e5c2a3fc070 // indirect
github.com/twmb/avro v1.8.0 // indirect
github.com/twmb/murmur3 v1.1.8 // indirect
github.com/twpayne/go-geom v1.6.1 // indirect
github.com/urfave/cli/v2 v2.27.7 // indirect
Expand Down
2 changes: 1 addition & 1 deletion public/bundle/free/go.mod
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -418,7 +418,7 @@ require (
github.com/spaolacci/murmur3 v1.1.0 // indirect
github.com/tilinna/z85 v1.0.0 // indirect
github.com/timandy/routine v1.1.5 // indirect
github.com/twmb/avro v1.7.3-0.20260513193503-1e5c2a3fc070 // indirect
github.com/twmb/avro v1.8.0 // indirect
github.com/twmb/go-cache v1.3.0 // indirect
github.com/twpayne/go-geom v1.6.1 // indirect
github.com/urfave/cli/v2 v2.27.7 // indirect
Expand Down
Loading