confluent: resolve named-type references in lame Avro unions too

The lame-union path -- raw_unions: false on schema_registry_decode, which
is the documented default -- carried the same bug as the raw path: string
branches like "Fee" in ["null", "Fee"] went through ecsAvroTypeToCommon
directly and collapsed to schema.Any, even when "Fee" was a previously-
defined record. The tagged-JSON envelope around each branch then wrapped
an Any inner, producing a structureless metadata tree.

Reroute the lame hydrator through the same ecsAvroResolveTypeRef helper
the raw path now uses, then re-apply the lame-specific wrapping
(tagged-Object envelope, type-name preserved as Common.Name to match the
wire-form tag). The non-Avro behavior of the lame envelope is unchanged;
only the inner Common is now correctly populated for name references.

This closes the same CON-468 bug class for the default-config path that
commit 4531c51 closed for the raw-unions path.

Author: Jeffail

internal/impl/confluent/ecs_avro.go

@@ -534,19 +534,14 @@ func ecsAvroHydrateRawUnion(cfg ecsAvroConfig, c *schema.Common, types []any) er
 func ecsAvroHydrateLameUnion(cfg ecsAvroConfig, c *schema.Common, types []any) error {
 	c.Type = schema.Union
 	for i, uObj := range types {
-		var childT schema.Common
-
-		switch ut := uObj.(type) {
-		case string:
-			childT = schema.Common{
-				Name: ut,
-				Type: ecsAvroTypeToCommon(ut),
-			}
-		case map[string]any:
-			var err error
-			if childT, err = ecsAvroFromAnyMap(cfg, ut); err != nil {
-				return fmt.Errorf("union `%v` child '%v': %w", c.Name, i, err)
-			}
+		childT, ok := ecsAvroResolveTypeRef(cfg, uObj)
+		if !ok {
+			return fmt.Errorf("union `%v` child '%v': could not resolve type %T", c.Name, i, uObj)
+		}
+		if s, isStr := uObj.(string); isStr {
+			// Lame-union children keep the type-name as the Common.Name so
+			// the tagged-JSON envelope key matches the Avro wire form.
+			childT.Name = s
 		}
 
 		if childT.Type == schema.Null {

internal/impl/confluent/ecs_avro_test.go

@@ -467,6 +467,55 @@ func TestEcsAvroRawUnionNullableRecordNested(t *testing.T) {
 	assert.Equal(t, "code", inner.Children[0].Name)
 }
 
+// TestEcsAvroLameUnionNameResolution covers the same CON-468 bug class in
+// the lame-union (rawUnion=false) path that the schema_registry_decode
+// processor takes by default. The lame envelope wraps each branch in a
+// tagged Object so the wire shape stays the same, but the inner Common
+// must still expand previously-defined named-type references rather than
+// collapsing them to schema.Any.
+func TestEcsAvroLameUnionNameResolution(t *testing.T) {
+	spec := []byte(`{
+		"type": "record",
+		"name": "Transfer",
+		"fields": [
+			{
+				"name": "primary_fee",
+				"type": {
+					"type": "record",
+					"name": "Fee",
+					"fields": [{"name": "amount", "type": "long"}]
+				}
+			},
+			{"name": "secondary_fee", "type": ["null", "Fee"]}
+		]
+	}`)
+	c, err := ecsAvroParseFromBytes(ecsAvroConfig{}, spec)
+	require.NoError(t, err)
+
+	secondary := c.Children[1]
+	assert.Equal(t, "secondary_fee", secondary.Name)
+	assert.Equal(t, schema.Union, secondary.Type)
+	require.Len(t, secondary.Children, 2)
+
+	// One child is the null branch (Common.Type=Null, Name="").
+	// The other is the tagged-Object envelope wrapping the resolved Fee.
+	var feeEnvelope schema.Common
+	for _, ch := range secondary.Children {
+		if ch.Type == schema.Object {
+			feeEnvelope = ch
+			break
+		}
+	}
+	require.Equal(t, schema.Object, feeEnvelope.Type, "Fee branch should be tagged-Object envelope")
+	require.Len(t, feeEnvelope.Children, 1)
+	feeInner := feeEnvelope.Children[0]
+	assert.Equal(t, "Fee", feeInner.Name)
+	assert.Equal(t, schema.Object, feeInner.Type, "name reference to Fee should resolve to its record structure, not schema.Any")
+	require.Len(t, feeInner.Children, 1)
+	assert.Equal(t, "amount", feeInner.Children[0].Name)
+	assert.Equal(t, schema.Int64, feeInner.Children[0].Type)
+}
+
 // TestEcsAvroRecordWithNilFields is a regression test for schemas where a
 // field's type is a record object without a "fields" key (e.g. back-reference
 // form {"type":"record","name":"Party"} or "fields":null from some generators).