From 9f3f4f0765ed67da0648abd41fe22efae92e0f1b Mon Sep 17 00:00:00 2001 From: Sagar Date: Thu, 30 Jul 2026 11:16:47 +0530 Subject: [PATCH 1/4] fix: convert numeric types inside array/map union branches in Avro serializer --- pkg/kafka/avro.go | 14 ++++++++++++++ 1 file changed, 14 insertions(+) diff --git a/pkg/kafka/avro.go b/pkg/kafka/avro.go index 5b9b1b08..3cabef17 100644 --- a/pkg/kafka/avro.go +++ b/pkg/kafka/avro.go @@ -360,6 +360,20 @@ func convertUnionField(fieldValue any, unionSchema *avro.UnionSchema) (any, erro continue } + // Arrays and maps are unnamed composite types: recurse into their + // elements so nested int/long fields get converted, and return the + // value unwrapped (hamba/avro resolves the branch by Go type). + if actualType.Type() == avro.Array || actualType.Type() == avro.Map { + if !isValueCompatibleWithSchema(fieldValue, actualType) { + continue + } + converted, err := convertFloat64ToIntForIntegerFields(fieldValue, actualType) + if err != nil { + continue + } + return converted, nil + } + // Named schemas (enums, fixed, records) need wrapping. if namedSchema, ok := actualType.(avro.NamedSchema); ok { if actualType.Type() == avro.Enum { From a7a5dac1522ee7f986673992fa0d85a298742847 Mon Sep 17 00:00:00 2001 From: Mostafa Moradian Date: Tue, 8 Sep 2026 13:00:26 +0200 Subject: [PATCH 2/4] fix: wrap map union branches and add array/map union branch tests - Wrap map branch values as {"map": value}: hamba/avro only encodes map[string]any union values in the type-name-wrapped form, while bare slices encode via the nullable-union path. The unwrapped map branch failed Serialize with "cannot encode union map with multiple entries". - Add tests for the #408 shape: ["null", array] branches with plain ["null", "int"] item fields (bare, wrapped, and null values), map branches (int values and map of records), and full AvroSerde.Serialize round-trips for both composite branch types. - Fix gofmt indentation and document first-match branch selection. --- pkg/kafka/avro.go | 33 ++-- pkg/kafka/avro_conversion_test.go | 293 ++++++++++++++++++++++++++++++ 2 files changed, 314 insertions(+), 12 deletions(-) diff --git a/pkg/kafka/avro.go b/pkg/kafka/avro.go index 3cabef17..82506599 100644 --- a/pkg/kafka/avro.go +++ b/pkg/kafka/avro.go @@ -361,18 +361,27 @@ func convertUnionField(fieldValue any, unionSchema *avro.UnionSchema) (any, erro } // Arrays and maps are unnamed composite types: recurse into their - // elements so nested int/long fields get converted, and return the - // value unwrapped (hamba/avro resolves the branch by Go type). - if actualType.Type() == avro.Array || actualType.Type() == avro.Map { - if !isValueCompatibleWithSchema(fieldValue, actualType) { - continue - } - converted, err := convertFloat64ToIntForIntegerFields(fieldValue, actualType) - if err != nil { - continue - } - return converted, nil - } + // elements so nested int/long fields get converted. + // As with primitives above, the first compatible branch wins, so + // ambiguous unions (e.g. ["null", array, array]) may + // pick the wrong branch. + if actualType.Type() == avro.Array || actualType.Type() == avro.Map { + if !isValueCompatibleWithSchema(fieldValue, actualType) { + continue + } + converted, err := convertFloat64ToIntForIntegerFields(fieldValue, actualType) + if err != nil { + continue + } + // Arrays are returned unwrapped: hamba/avro encodes a bare slice + // with the non-null branch of a nullable union. Maps must be + // wrapped by type name: hamba/avro only encodes map[string]any + // union values in the {"map": value} form. + if actualType.Type() == avro.Map { + return map[string]any{string(avro.Map): converted}, nil + } + return converted, nil + } // Named schemas (enums, fixed, records) need wrapping. if namedSchema, ok := actualType.(avro.NamedSchema); ok { diff --git a/pkg/kafka/avro_conversion_test.go b/pkg/kafka/avro_conversion_test.go index 2a9738bc..8bbe28b6 100644 --- a/pkg/kafka/avro_conversion_test.go +++ b/pkg/kafka/avro_conversion_test.go @@ -451,6 +451,132 @@ func TestConvertUnionField_UnknownWrappedPrimitiveKey(t *testing.T) { assert.Equal(t, data, got) } +// TestConvertUnionField_ArrayBranchWithIntUnionItems covers the issue #408 +// shape: a union whose non-null branch is an array of records containing a +// plain ["null", "int"] union field. The array branch must be recursed into +// so the nested float64 values become int32, and the value must be returned +// unwrapped (hamba/avro resolves unnamed composite branches by Go type). +func TestConvertUnionField_ArrayBranchWithIntUnionItems(t *testing.T) { + schemaJSON := `["null", { + "type": "array", + "items": { + "type": "record", + "name": "Line", + "namespace": "com.example", + "fields": [ + {"name": "lineNumber", "type": ["null", "int"], "default": null} + ] + } + }]` + schema, err := avro.Parse(schemaJSON) + require.NoError(t, err) + unionSchema := schema.(*avro.UnionSchema) + + tests := []struct { + name string + data any + want any + }{ + { + name: "bare int value", + data: []any{ + map[string]any{"lineNumber": float64(1)}, + }, + want: []any{ + map[string]any{"lineNumber": int32(1)}, + }, + }, + { + name: "wrapped int value", + data: []any{ + map[string]any{"lineNumber": map[string]any{"int": float64(1)}}, + }, + want: []any{ + map[string]any{"lineNumber": int32(1)}, + }, + }, + { + name: "null int value", + data: []any{ + map[string]any{"lineNumber": nil}, + }, + want: []any{ + map[string]any{"lineNumber": nil}, + }, + }, + } + + for _, testCase := range tests { + t.Run(testCase.name, func(t *testing.T) { + got, err := convertUnionField(testCase.data, unionSchema) + assert.NoError(t, err) + assert.Equal(t, testCase.want, got) + }) + } +} + +// TestConvertUnionField_MapBranch covers unions whose non-null branch is a +// map: the map branch must be recursed into so nested float64 values become +// int32/int64, and the value must be returned wrapped as {"map": value} +// (hamba/avro only encodes map[string]any union values in that form). +func TestConvertUnionField_MapBranch(t *testing.T) { + tests := []struct { + name string + schemaJSON string + data any + want any + }{ + { + name: "map with int values", + schemaJSON: `["null", {"type": "map", "values": "int"}]`, + data: map[string]any{ + "key1": float64(1), + "key2": float64(2), + }, + want: map[string]any{ + "map": map[string]any{ + "key1": int32(1), + "key2": int32(2), + }, + }, + }, + { + name: "map of records with int unions", + schemaJSON: `["null", { + "type": "map", + "values": { + "type": "record", + "name": "Line", + "namespace": "com.example", + "fields": [ + {"name": "lineNumber", "type": ["null", "int"], "default": null} + ] + } + }]`, + data: map[string]any{ + "line1": map[string]any{"lineNumber": float64(1)}, + }, + want: map[string]any{ + "map": map[string]any{ + "line1": map[string]any{"lineNumber": int32(1)}, + }, + }, + }, + } + + for _, testCase := range tests { + t.Run(testCase.name, func(t *testing.T) { + schema, err := avro.Parse(testCase.schemaJSON) + require.NoError(t, err) + unionSchema := schema.(*avro.UnionSchema) + + got, err := convertUnionField(testCase.data, unionSchema) + assert.NoError(t, err) + assert.Equal(t, testCase.want, got) + }) + } +} + // TestConvertFloat64ToIntForIntegerFields_UnionWithLogicalType tests the exact scenario from issue #376 // where a union type contains an int with logical type "date". func TestConvertFloat64ToIntForIntegerFields_UnionWithLogicalType(t *testing.T) { @@ -944,6 +1070,173 @@ func TestSerializeDeserializeRoundTrip_WithUnions(t *testing.T) { assert.Equal(t, []byte{1, 2, 3}, result["bytesField"]) } +// TestSerializeDeserializeRoundTrip_UnionArrayBranch covers the exact issue +// #408 shape: a ["null", array] union branch whose items contain a +// plain ["null", "int"] union field. Before the fix, both the bare and the +// wrapped form failed to serialize ("avro: unknown union type double" / +// "avro: float64 is unsupported for Avro int"). +func TestSerializeDeserializeRoundTrip_UnionArrayBranch(t *testing.T) { + schemaJSON := `{ + "type": "record", + "name": "Order", + "namespace": "com.example", + "fields": [ + { + "name": "lines", + "type": ["null", { + "type": "array", + "items": { + "type": "record", + "name": "Line", + "fields": [ + {"name": "lineNumber", "type": ["null", "int"], "default": null} + ] + } + }], + "default": null + } + ] + }` + avroSerde := &AvroSerde{} + schema := &Schema{ + ID: 408, + Schema: schemaJSON, + Version: 1, + Subject: "issue-408-roundtrip", + } + + tests := []struct { + name string + data map[string]any + }{ + { + name: "bare int value", + data: map[string]any{ + "lines": []any{ + map[string]any{"lineNumber": float64(1)}, + }, + }, + }, + { + name: "wrapped int value", + data: map[string]any{ + "lines": []any{ + map[string]any{"lineNumber": map[string]any{"int": float64(1)}}, + }, + }, + }, + { + name: "null lines", + data: map[string]any{"lines": nil}, + }, + } + + for _, testCase := range tests { + t.Run(testCase.name, func(t *testing.T) { + serialized, serdeErr := avroSerde.Serialize(testCase.data, schema) + require.Nil(t, serdeErr, "Serialize should not return error") + require.NotNil(t, serialized, "Serialized data should not be nil") + + deserialized, deserErr := avroSerde.Deserialize(serialized, schema) + require.Nil(t, deserErr, "Deserialize should not return error") + require.NotNil(t, deserialized, "Deserialized data should not be nil") + + result := deserialized.(map[string]any) + if testCase.data["lines"] == nil { + assert.Nil(t, result["lines"]) + return + } + + // hamba/avro decodes unnamed composite union branches (array/map) + // into a type-name-wrapped map; accept that form until decode-side + // unwrapping is addressed separately. + linesValue := result["lines"] + if envelope, ok := linesValue.(map[string]any); ok { + linesValue = envelope["array"] + } + lines, ok := linesValue.([]any) + require.True(t, ok, "lines should be an array, got %T", result["lines"]) + require.Len(t, lines, 1) + line, ok := lines[0].(map[string]any) + require.True(t, ok, "line should be a record, got %T", lines[0]) + + // After deserialization, int values may be returned as int (platform-dependent) + switch lineNumber := line["lineNumber"].(type) { + case int32: + assert.Equal(t, int32(1), lineNumber) + case int: + assert.Equal(t, 1, lineNumber) + default: + t.Fatalf("unexpected type for lineNumber: %T", line["lineNumber"]) + } + }) + } +} + +// TestSerializeDeserializeRoundTrip_UnionMapBranch verifies that a +// ["null", map] union branch serializes from bare float64 values. +func TestSerializeDeserializeRoundTrip_UnionMapBranch(t *testing.T) { + schemaJSON := `{ + "type": "record", + "name": "Inventory", + "namespace": "com.example", + "fields": [ + { + "name": "counts", + "type": ["null", {"type": "map", "values": "int"}], + "default": null + } + ] + }` + avroSerde := &AvroSerde{} + schema := &Schema{ + ID: 408, + Schema: schemaJSON, + Version: 1, + Subject: "issue-408-map-roundtrip", + } + + originalData := map[string]any{ + "counts": map[string]any{ + "key1": float64(1), + "key2": float64(2), + }, + } + + serialized, serdeErr := avroSerde.Serialize(originalData, schema) + require.Nil(t, serdeErr, "Serialize should not return error") + require.NotNil(t, serialized, "Serialized data should not be nil") + + deserialized, deserErr := avroSerde.Deserialize(serialized, schema) + require.Nil(t, deserErr, "Deserialize should not return error") + require.NotNil(t, deserialized, "Deserialized data should not be nil") + + result := deserialized.(map[string]any) + // hamba/avro decodes unnamed composite union branches (array/map) into a + // type-name-wrapped map; accept that form until decode-side unwrapping is + // addressed separately. + countsValue := result["counts"] + if envelope, ok := countsValue.(map[string]any); ok { + if wrapped, exists := envelope["map"]; exists { + countsValue = wrapped + } + } + counts, ok := countsValue.(map[string]any) + require.True(t, ok, "counts should be a map, got %T", result["counts"]) + require.Len(t, counts, 2) + for key, want := range map[string]int32{"key1": 1, "key2": 2} { + // After deserialization, int values may be returned as int (platform-dependent) + switch value := counts[key].(type) { + case int32: + assert.Equal(t, want, value) + case int: + assert.Equal(t, int(want), value) + default: + t.Fatalf("unexpected type for counts[%q]: %T", key, counts[key]) + } + } +} + func TestAvroMarshal_UnionLogicalTypeDateAcceptedShapes(t *testing.T) { schema, err := avro.Parse(testDocumentLogicalDateSchemaJSON) require.NoError(t, err) From bfba64152824bdf8ce7ad3dd05d93101fa1025e8 Mon Sep 17 00:00:00 2001 From: Mostafa Moradian Date: Tue, 8 Sep 2026 13:00:27 +0200 Subject: [PATCH 3/4] chore(lint): disable exhaustruct/exhaustruct_v5 --- .golangci.yaml | 1 + 1 file changed, 1 insertion(+) diff --git a/.golangci.yaml b/.golangci.yaml index a479bd7c..4c314e13 100644 --- a/.golangci.yaml +++ b/.golangci.yaml @@ -5,6 +5,7 @@ linters: - cyclop - depguard - exhaustruct + - exhaustruct_v5 - forcetypeassert - funlen - gochecknoglobals From 24158dc8fef0b1746935993cf8f06718a1c75667 Mon Sep 17 00:00:00 2001 From: Mostafa Moradian Date: Tue, 8 Sep 2026 13:04:46 +0200 Subject: [PATCH 4/4] ci: upgrade golangci-lint to v2.13.2 v2.12.2 does not recognize the exhaustruct_v5 linter name (renamed in v2.13.x), failing config validation. v2.13.2 matches the version mise installs locally (golangci-lint = "2"), keeping CI and local lint in sync. --- .github/workflows/test.yaml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.github/workflows/test.yaml b/.github/workflows/test.yaml index b7178451..652a78c1 100644 --- a/.github/workflows/test.yaml +++ b/.github/workflows/test.yaml @@ -45,7 +45,7 @@ jobs: - name: Lint code issues 🚨 uses: golangci/golangci-lint-action@1e7e51e771db61008b38414a730f564565cf7c20 # v9.2.0 with: - version: v2.12.2 + version: v2.13.2 - name: Validate declarations 📐 working-directory: api-docs