Fix redefining a named Avro type in a diamond dependency pattern (#2238)

rayokota · web-flow · commit f01ea645a59f · 2026-04-28T10:51:30.000-07:00
* Fix redefining a named Avro type in a diamond dependency pattern

* Fix style

* Update changelog
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -6,6 +6,7 @@
 
 - Fix URL joining in Python client (#2228)
 - Handle anyOf/allOf in JSON transforms (#2237)
+- Fix redefining a named Avro type in a diamond dependency pattern (#2238)
 
 
 ## v2.14.0 - 2026-04-01
diff --git a/src/confluent_kafka/schema_registry/_async/avro.py b/src/confluent_kafka/schema_registry/_async/avro.py
@@ -63,7 +63,7 @@ async def _resolve_named_schema(
     :param schema_registry_client: SchemaRegistryClient to use for retrieval.
     :return: named_schemas dict.
     """
-    named_schemas = {}
+    named_schemas: Dict[str, AvroSchema] = {}
     if schema.references is not None:
         for ref in schema.references:
             if ref.subject is None or ref.version is None:
@@ -72,18 +72,26 @@ async def _resolve_named_schema(
             ref_named_schemas = await _resolve_named_schema(referenced_schema.schema, schema_registry_client)
             if referenced_schema.schema.schema_str is None:
                 raise TypeError("Schema string cannot be None")
-
-            parsed_schema = parse_schema_with_repo(referenced_schema.schema.schema_str, named_schemas=ref_named_schemas)
-            named_schemas.update(ref_named_schemas)
             if ref.name is None:
                 raise TypeError("Name cannot be None")
-            named_schemas[ref.name] = parsed_schema
+            named_schemas.update(ref_named_schemas)
+            # Store the raw (unparsed) schema dict. Pre-parsing here would inline
+            # any sub-references inside this schema; if the same sub-reference is
+            # also reachable through another sibling reference (a "diamond"
+            # dependency), the top-level load_schema would then inject duplicate
+            # inline definitions and fail with "redefined named type". Keeping
+            # the raw form lets the top-level load_schema resolve every named
+            # type exactly once.
+            raw_schema = json.loads(referenced_schema.schema.schema_str)
+            named_schemas[ref.name] = raw_schema
             # Also store under fully-qualified name so fastavro can resolve
             # namespace-qualified type references
-            if isinstance(parsed_schema, dict) and 'name' in parsed_schema:
-                fqn = parsed_schema['name']
+            if isinstance(raw_schema, dict) and 'name' in raw_schema:
+                ns = raw_schema.get('namespace')
+                name = raw_schema['name']
+                fqn = f"{ns}.{name}" if ns and '.' not in name else name
                 if fqn != ref.name:
-                    named_schemas[fqn] = parsed_schema
+                    named_schemas[fqn] = raw_schema
     return named_schemas
 
 
diff --git a/src/confluent_kafka/schema_registry/_sync/avro.py b/src/confluent_kafka/schema_registry/_sync/avro.py
@@ -60,7 +60,7 @@ def _resolve_named_schema(schema: Schema, schema_registry_client: SchemaRegistry
     :param schema_registry_client: SchemaRegistryClient to use for retrieval.
     :return: named_schemas dict.
     """
-    named_schemas = {}
+    named_schemas: Dict[str, AvroSchema] = {}
     if schema.references is not None:
         for ref in schema.references:
             if ref.subject is None or ref.version is None:
@@ -69,18 +69,26 @@ def _resolve_named_schema(schema: Schema, schema_registry_client: SchemaRegistry
             ref_named_schemas = _resolve_named_schema(referenced_schema.schema, schema_registry_client)
             if referenced_schema.schema.schema_str is None:
                 raise TypeError("Schema string cannot be None")
-
-            parsed_schema = parse_schema_with_repo(referenced_schema.schema.schema_str, named_schemas=ref_named_schemas)
-            named_schemas.update(ref_named_schemas)
             if ref.name is None:
                 raise TypeError("Name cannot be None")
-            named_schemas[ref.name] = parsed_schema
+            named_schemas.update(ref_named_schemas)
+            # Store the raw (unparsed) schema dict. Pre-parsing here would inline
+            # any sub-references inside this schema; if the same sub-reference is
+            # also reachable through another sibling reference (a "diamond"
+            # dependency), the top-level load_schema would then inject duplicate
+            # inline definitions and fail with "redefined named type". Keeping
+            # the raw form lets the top-level load_schema resolve every named
+            # type exactly once.
+            raw_schema = json.loads(referenced_schema.schema.schema_str)
+            named_schemas[ref.name] = raw_schema
             # Also store under fully-qualified name so fastavro can resolve
             # namespace-qualified type references
-            if isinstance(parsed_schema, dict) and 'name' in parsed_schema:
-                fqn = parsed_schema['name']
+            if isinstance(raw_schema, dict) and 'name' in raw_schema:
+                ns = raw_schema.get('namespace')
+                name = raw_schema['name']
+                fqn = f"{ns}.{name}" if ns and '.' not in name else name
                 if fqn != ref.name:
-                    named_schemas[fqn] = parsed_schema
+                    named_schemas[fqn] = raw_schema
     return named_schemas
 
 
diff --git a/tests/schema_registry/_async/test_avro_serdes.py b/tests/schema_registry/_async/test_avro_serdes.py
@@ -467,6 +467,86 @@ async def test_avro_serialize_union_with_references():
     assert obj == obj2
 
 
+async def test_avro_diamond_dependency_references():
+    # Two sibling references (OrderDetails, InvoiceDetails) both depend on the
+    # same named type (Address). Without the fix in _resolve_named_schema, each
+    # branch is pre-parsed with Address inlined, and the top-level parse then
+    # raises SchemaParseException("redefined named type ...Address").
+    conf = {'url': _BASE_URL}
+    client = AsyncSchemaRegistryClient.new_client(conf)
+    ser_conf = {'auto.register.schemas': False, 'use.latest.version': True}
+
+    ns = "com.example.diamond"
+    address_schema = {
+        'type': 'record',
+        'name': 'Address',
+        'namespace': ns,
+        'fields': [{'name': 'street', 'type': 'string'}],
+    }
+    order_schema = {
+        'type': 'record',
+        'name': 'OrderDetails',
+        'namespace': ns,
+        'fields': [{'name': 'shipping_address', 'type': f'{ns}.Address'}],
+    }
+    invoice_schema = {
+        'type': 'record',
+        'name': 'InvoiceDetails',
+        'namespace': ns,
+        'fields': [{'name': 'billing_address', 'type': f'{ns}.Address'}],
+    }
+    root_schema = {
+        'type': 'record',
+        'name': 'OrderEvent',
+        'namespace': ns,
+        'fields': [
+            {'name': 'order', 'type': f'{ns}.OrderDetails'},
+            {'name': 'invoice', 'type': f'{ns}.InvoiceDetails'},
+        ],
+    }
+
+    await client.register_schema('diamond-Address', Schema(json.dumps(address_schema)))
+    await client.register_schema(
+        'diamond-OrderDetails',
+        Schema(
+            json.dumps(order_schema),
+            'AVRO',
+            [SchemaReference(f'{ns}.Address', 'diamond-Address', 1)],
+        ),
+    )
+    await client.register_schema(
+        'diamond-InvoiceDetails',
+        Schema(
+            json.dumps(invoice_schema),
+            'AVRO',
+            [SchemaReference(f'{ns}.Address', 'diamond-Address', 1)],
+        ),
+    )
+    await client.register_schema(
+        _SUBJECT,
+        Schema(
+            json.dumps(root_schema),
+            'AVRO',
+            [
+                SchemaReference(f'{ns}.OrderDetails', 'diamond-OrderDetails', 1),
+                SchemaReference(f'{ns}.InvoiceDetails', 'diamond-InvoiceDetails', 1),
+            ],
+        ),
+    )
+
+    obj = {
+        'order': {'shipping_address': {'street': '123 Main St'}},
+        'invoice': {'billing_address': {'street': '456 Elm St'}},
+    }
+    ser = await AsyncAvroSerializer(client, schema_str=None, conf=ser_conf)
+    ser_ctx = SerializationContext(_TOPIC, MessageField.VALUE)
+    obj_bytes = await ser(obj, ser_ctx)
+
+    deser = await AsyncAvroDeserializer(client)
+    obj2 = await deser(obj_bytes, ser_ctx)
+    assert obj == obj2
+
+
 async def test_avro_schema_evolution():
     conf = {'url': _BASE_URL}
     client = AsyncSchemaRegistryClient.new_client(conf)
diff --git a/tests/schema_registry/_sync/test_avro_serdes.py b/tests/schema_registry/_sync/test_avro_serdes.py
@@ -467,6 +467,86 @@ def test_avro_serialize_union_with_references():
     assert obj == obj2
 
 
+def test_avro_diamond_dependency_references():
+    # Two sibling references (OrderDetails, InvoiceDetails) both depend on the
+    # same named type (Address). Without the fix in _resolve_named_schema, each
+    # branch is pre-parsed with Address inlined, and the top-level parse then
+    # raises SchemaParseException("redefined named type ...Address").
+    conf = {'url': _BASE_URL}
+    client = SchemaRegistryClient.new_client(conf)
+    ser_conf = {'auto.register.schemas': False, 'use.latest.version': True}
+
+    ns = "com.example.diamond"
+    address_schema = {
+        'type': 'record',
+        'name': 'Address',
+        'namespace': ns,
+        'fields': [{'name': 'street', 'type': 'string'}],
+    }
+    order_schema = {
+        'type': 'record',
+        'name': 'OrderDetails',
+        'namespace': ns,
+        'fields': [{'name': 'shipping_address', 'type': f'{ns}.Address'}],
+    }
+    invoice_schema = {
+        'type': 'record',
+        'name': 'InvoiceDetails',
+        'namespace': ns,
+        'fields': [{'name': 'billing_address', 'type': f'{ns}.Address'}],
+    }
+    root_schema = {
+        'type': 'record',
+        'name': 'OrderEvent',
+        'namespace': ns,
+        'fields': [
+            {'name': 'order', 'type': f'{ns}.OrderDetails'},
+            {'name': 'invoice', 'type': f'{ns}.InvoiceDetails'},
+        ],
+    }
+
+    client.register_schema('diamond-Address', Schema(json.dumps(address_schema)))
+    client.register_schema(
+        'diamond-OrderDetails',
+        Schema(
+            json.dumps(order_schema),
+            'AVRO',
+            [SchemaReference(f'{ns}.Address', 'diamond-Address', 1)],
+        ),
+    )
+    client.register_schema(
+        'diamond-InvoiceDetails',
+        Schema(
+            json.dumps(invoice_schema),
+            'AVRO',
+            [SchemaReference(f'{ns}.Address', 'diamond-Address', 1)],
+        ),
+    )
+    client.register_schema(
+        _SUBJECT,
+        Schema(
+            json.dumps(root_schema),
+            'AVRO',
+            [
+                SchemaReference(f'{ns}.OrderDetails', 'diamond-OrderDetails', 1),
+                SchemaReference(f'{ns}.InvoiceDetails', 'diamond-InvoiceDetails', 1),
+            ],
+        ),
+    )
+
+    obj = {
+        'order': {'shipping_address': {'street': '123 Main St'}},
+        'invoice': {'billing_address': {'street': '456 Elm St'}},
+    }
+    ser = AvroSerializer(client, schema_str=None, conf=ser_conf)
+    ser_ctx = SerializationContext(_TOPIC, MessageField.VALUE)
+    obj_bytes = ser(obj, ser_ctx)
+
+    deser = AvroDeserializer(client)
+    obj2 = deser(obj_bytes, ser_ctx)
+    assert obj == obj2
+
+
 def test_avro_schema_evolution():
     conf = {'url': _BASE_URL}
     client = SchemaRegistryClient.new_client(conf)
