[fix][broker] Keep accepting Avro named type references written as objects

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/AvroSchemaBasedCompatibilityCheck.java

@@ -52,14 +52,10 @@ public void checkCompatible(Iterable<SchemaData> from, SchemaData to, SchemaComp
         checkArgument(from != null, "check compatibility list is null");
         try {
             for (SchemaData schemaData : from) {
-                Schema.Parser parser =
-                        new Schema.Parser(StructSchemaDataValidator.COMPATIBLE_NAME_VALIDATOR);
-                parser.setValidateDefaults(false);
-                fromList.addFirst(parser.parse(new String(schemaData.getData(), UTF_8)));
+                fromList.addFirst(StructSchemaDataValidator.parseAvroSchema(
+                        new String(schemaData.getData(), UTF_8), false));
             }
-            Schema.Parser parser = new Schema.Parser(StructSchemaDataValidator.COMPATIBLE_NAME_VALIDATOR);
-            parser.setValidateDefaults(false);
-            Schema toSchema = parser.parse(new String(to.getData(), UTF_8));
+            Schema toSchema = StructSchemaDataValidator.parseAvroSchema(new String(to.getData(), UTF_8), false);
             SchemaValidator schemaValidator = createSchemaValidator(strategy);
             schemaValidator.validate(toSchema, fromList);
         } catch (SchemaParseException e) {

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/JsonSchemaCompatibilityCheck.java

@@ -22,7 +22,6 @@
 import com.fasterxml.jackson.databind.ObjectReader;
 import com.fasterxml.jackson.module.jsonSchema.JsonSchema;
 import java.io.IOException;
-import org.apache.avro.Schema;
 import org.apache.pulsar.broker.service.schema.exceptions.IncompatibleSchemaException;
 import org.apache.pulsar.broker.service.schema.validator.StructSchemaDataValidator;
 import org.apache.pulsar.common.policies.data.SchemaCompatibilityStrategy;
@@ -106,10 +105,7 @@ private void isCompatibleJsonSchema(SchemaData from, SchemaData to) throws Incom
 
     private boolean isAvroSchema(SchemaData schemaData) {
         try {
-
-            Schema.Parser fromParser = new Schema.Parser(StructSchemaDataValidator.COMPATIBLE_NAME_VALIDATOR);
-            fromParser.setValidateDefaults(false);
-            Schema fromSchema = fromParser.parse(new String(schemaData.getData(), UTF_8));
+            StructSchemaDataValidator.parseAvroSchema(new String(schemaData.getData(), UTF_8), false);
             return true;
         } catch (Exception e) {
             return false;

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/SchemaRegistryServiceImpl.java

@@ -403,14 +403,13 @@ public CompletableFuture<SchemaVersion> getSchemaVersionBySchemaData(
         final CompletableFuture<SchemaVersion> completableFuture = new CompletableFuture<>();
         SchemaVersion schemaVersion;
         if (isUsingAvroSchemaParser(schemaData.getType())) {
-            Schema.Parser parser = new Schema.Parser(StructSchemaDataValidator.COMPATIBLE_NAME_VALIDATOR);
-            Schema newSchema = parser.parse(new String(schemaData.getData(), UTF_8));
+            Schema newSchema = StructSchemaDataValidator.parseAvroSchema(
+                    new String(schemaData.getData(), UTF_8), true);
 
             for (SchemaAndMetadata schemaAndMetadata : schemaAndMetadataList) {
                 if (isUsingAvroSchemaParser(schemaAndMetadata.schema.getType())) {
-                    Schema.Parser existParser =
-                            new Schema.Parser(StructSchemaDataValidator.COMPATIBLE_NAME_VALIDATOR);
-                    Schema existSchema = existParser.parse(new String(schemaAndMetadata.schema.getData(), UTF_8));
+                    Schema existSchema = StructSchemaDataValidator.parseAvroSchema(
+                            new String(schemaAndMetadata.schema.getData(), UTF_8), true);
                     if (newSchema.equals(existSchema) && schemaAndMetadata.schema.getType() == schemaData.getType()) {
                         schemaVersion = schemaAndMetadata.version;
                         completableFuture.complete(schemaVersion);

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/validator/StructSchemaDataValidator.java

@@ -26,6 +26,7 @@
 import org.apache.avro.Schema;
 import org.apache.pulsar.broker.service.schema.exceptions.InvalidSchemaDataException;
 import org.apache.pulsar.common.protocol.schema.SchemaData;
+import org.apache.pulsar.common.schema.AvroSchemaCompat;
 import org.apache.pulsar.common.schema.SchemaType;
 import org.apache.pulsar.common.util.ObjectMapperFactory;
 
@@ -61,9 +62,7 @@ public void validate(SchemaData schemaData) throws InvalidSchemaDataException {
         byte[] data = schemaData.getData();
 
         try {
-            Schema.Parser avroSchemaParser = new Schema.Parser(COMPATIBLE_NAME_VALIDATOR);
-            avroSchemaParser.setValidateDefaults(false);
-            Schema schema = avroSchemaParser.parse(new String(data, UTF_8));
+            Schema schema = parseAvroSchema(new String(data, UTF_8), false);
             if (SchemaType.AVRO.equals(schemaData.getType())) {
                 checkAvroSchemaTypeSupported(schema);
             }
@@ -86,6 +85,20 @@ public void validate(SchemaData schemaData) throws InvalidSchemaDataException {
         }
     }
 
+    /**
+     * Parse an Avro schema definition the way the broker accepts it: with the compatible name validator and
+     * with named type references in the pre Avro 1.12.2 object form.
+     *
+     * @param schemaDefinition the schema definition as JSON
+     * @param validateDefaults whether to validate the default values of the schema
+     * @return the parsed schema
+     */
+    public static Schema parseAvroSchema(String schemaDefinition, boolean validateDefaults) {
+        Schema.Parser parser = new Schema.Parser(COMPATIBLE_NAME_VALIDATOR);
+        parser.setValidateDefaults(validateDefaults);
+        return parser.parse(AvroSchemaCompat.normalizeNamedTypeReferences(schemaDefinition));
+    }
+
     static void checkAvroSchemaTypeSupported(Schema schema) throws InvalidSchemaDataException {
             switch (schema.getType()) {
                 case RECORD: {

pulsar-broker/src/test/java/org/apache/pulsar/broker/service/schema/BaseAvroSchemaCompatibilityTest.java

@@ -152,6 +152,26 @@ public void testForwardCompatibility() {
      * Make sure the new schema is forward- and
      * backward-compatible from the latest to newest and from the newest to latest.
      */
+    @Test
+    public void testLegacyNamedTypeReferenceForm() {
+        // Avro 1.12.2 rejects named type references written as {"type": "name"} (AVRO-4176); schemas stored in
+        // that form must remain compatible with the same schema written with bare names
+        String colorEnum = "{\"type\":\"enum\",\"name\":\"Color\",\"namespace\":\"org.example.shapes\","
+                + "\"symbols\":[\"RED\",\"BLUE\"]}";
+        String legacyForm = "{\"type\":\"record\",\"name\":\"Drawing\",\"namespace\":\"org.example.shapes\","
+                + "\"fields\":[{\"name\":\"background\",\"type\":" + colorEnum + "},"
+                + "{\"name\":\"outline\",\"type\":{\"type\":\"org.example.shapes.Color\"}}]}";
+        String bareNameForm = legacyForm.replace("{\"type\":\"org.example.shapes.Color\"}",
+                "\"org.example.shapes.Color\"");
+        SchemaCompatibilityCheck schemaCompatibilityCheck = getSchemaCheck();
+        Assert.assertTrue(schemaCompatibilityCheck.isCompatible(getSchemaData(legacyForm),
+                        getSchemaData(bareNameForm), SchemaCompatibilityStrategy.FULL),
+                "the legacy and the bare name form of a schema are the same schema");
+        Assert.assertTrue(schemaCompatibilityCheck.isCompatible(getSchemaData(bareNameForm),
+                        getSchemaData(legacyForm), SchemaCompatibilityStrategy.FULL),
+                "the legacy and the bare name form of a schema are the same schema");
+    }
+
     @Test
     public void testFullCompatibility() {
         SchemaCompatibilityCheck schemaCompatibilityCheck = getSchemaCheck();

pulsar-broker/src/test/java/org/apache/pulsar/broker/service/schema/validator/SchemaDataValidatorTest.java

@@ -368,4 +368,37 @@ public void testAvroCompatible() throws InvalidSchemaDataException {
         StructSchemaDataValidator.of().validate(SchemaData.fromSchemaInfo(protobufSchema.getSchemaInfo()));
     }
 
+    // Avro 1.12.2 rejects named type references written as {"type": "name"} (AVRO-4176); such schemas were
+    // accepted before and must still be accepted
+
+    @DataProvider(name = "legacyNamedTypeReferenceSchemas")
+    public Object[][] legacyNamedTypeReferenceSchemas() {
+        String colorEnum = "{\"type\":\"enum\",\"name\":\"Color\",\"namespace\":\"org.example.shapes\","
+                + "\"symbols\":[\"RED\",\"BLUE\"]}";
+        String drawing = "{\"type\":\"record\",\"name\":\"Drawing\",\"namespace\":\"org.example.shapes\","
+                + "\"fields\":[{\"name\":\"background\",\"type\":" + colorEnum + "},"
+                + "{\"name\":\"outline\",\"type\":{\"type\":\"org.example.shapes.Color\"}},"
+                + "{\"name\":\"palette\",\"type\":{\"type\":\"array\","
+                + "\"items\":{\"type\":\"org.example.shapes.Color\"}}}]}";
+        return new Object[][] {
+            { SchemaType.AVRO, drawing },
+            { SchemaType.JSON, drawing },
+        };
+    }
+
+    @Test(dataProvider = "legacyNamedTypeReferenceSchemas")
+    public void testLegacyNamedTypeReferenceIsAccepted(SchemaType type, String schemaDefinition)
+            throws InvalidSchemaDataException {
+        SchemaData data = SchemaData.builder()
+            .type(type)
+            .data(schemaDefinition.getBytes(UTF_8))
+            .build();
+        SchemaDataValidator.validateSchemaData(data, false);
+
+        org.apache.avro.Schema parsed = StructSchemaDataValidator.parseAvroSchema(schemaDefinition, false);
+        Assert.assertEquals(parsed.getField("outline").schema().getFullName(), "org.example.shapes.Color");
+        Assert.assertEquals(parsed.getField("palette").schema().getElementType().getFullName(),
+                "org.example.shapes.Color");
+    }
+
 }

pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/generic/GenericJsonRecord.java

@@ -26,6 +26,7 @@
 import java.util.stream.Collectors;
 import lombok.CustomLog;
 import org.apache.pulsar.client.api.schema.Field;
+import org.apache.pulsar.common.schema.AvroSchemaCompat;
 import org.apache.pulsar.common.schema.SchemaInfo;
 import org.apache.pulsar.common.schema.SchemaType;
 import org.apache.pulsar.common.util.ObjectMapperFactory;
@@ -130,7 +131,7 @@ private boolean isBinaryValue(String fieldName) {
     private static org.apache.avro.Schema parseAvroSchema(String schemaJson) {
         final org.apache.avro.Schema.Parser parser = new org.apache.avro.Schema.Parser();
         parser.setValidateDefaults(false);
-        return parser.parse(schemaJson);
+        return parser.parse(AvroSchemaCompat.normalizeNamedTypeReferences(schemaJson));
     }
 
     @Override

pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/util/SchemaUtil.java

@@ -29,6 +29,7 @@
 import org.apache.pulsar.client.impl.schema.AvroSchema;
 import org.apache.pulsar.client.impl.schema.SchemaDefinitionBuilderImpl;
 import org.apache.pulsar.client.impl.schema.SchemaInfoImpl;
+import org.apache.pulsar.common.schema.AvroSchemaCompat;
 import org.apache.pulsar.common.schema.SchemaInfo;
 import org.apache.pulsar.common.schema.SchemaType;
 
@@ -54,7 +55,7 @@ public static boolean getJsr310ConversionEnabled(SchemaInfo schemaInfo) {
     public static Schema parseAvroSchema(String schemaJson) {
         final Schema.Parser parser = new Schema.Parser(NameValidator.NO_VALIDATION);
         parser.setValidateDefaults(false);
-        return parser.parse(schemaJson);
+        return parser.parse(AvroSchemaCompat.normalizeNamedTypeReferences(schemaJson));
     }
 
     public static <T> SchemaInfo parseSchemaInfo(SchemaDefinition<T> schemaDefinition, SchemaType schemaType) {

pulsar-client/src/test/java/org/apache/pulsar/client/impl/schema/util/SchemaUtilTest.java

@@ -19,6 +19,7 @@
 package org.apache.pulsar.client.impl.schema.util;
 
 import static org.apache.pulsar.client.impl.schema.SchemaDefinitionBuilderImpl.JSR310_CONVERSION_ENABLED;
+import static org.testng.Assert.assertEquals;
 import static org.testng.Assert.assertFalse;
 import static org.testng.Assert.assertTrue;
 import java.util.HashMap;
@@ -108,4 +109,22 @@ private static SchemaInfo enabledJsr310PropertiesSchema() {
                 .build();
     }
 
+    @Test
+    public void testParseAvroSchemaWithLegacyNamedTypeReference() {
+        // Avro 1.12.2 rejects named type references written as {"type": "name"} (AVRO-4176); schemas stored in
+        // that form must still be readable
+        String colorEnum = "{\"type\":\"enum\",\"name\":\"Color\",\"namespace\":\"org.example.shapes\","
+                + "\"symbols\":[\"RED\",\"BLUE\"]}";
+        String legacyForm = "{\"type\":\"record\",\"name\":\"Drawing\",\"namespace\":\"org.example.shapes\","
+                + "\"fields\":[{\"name\":\"background\",\"type\":" + colorEnum + "},"
+                + "{\"name\":\"outline\",\"type\":{\"type\":\"org.example.shapes.Color\"}},"
+                + "{\"name\":\"highlight\",\"type\":[\"null\",{\"type\":\"org.example.shapes.Color\"}],"
+                + "\"default\":null}]}";
+        String bareNameForm = legacyForm.replace("{\"type\":\"org.example.shapes.Color\"}",
+                "\"org.example.shapes.Color\"");
+
+        org.apache.avro.Schema parsed = SchemaUtil.parseAvroSchema(legacyForm);
+        assertEquals(parsed, SchemaUtil.parseAvroSchema(bareNameForm));
+        assertEquals(parsed.getField("outline").schema().getFullName(), "org.example.shapes.Color");
+    }
 }

pulsar-common/src/main/java/org/apache/pulsar/common/schema/AvroSchemaCompat.java

@@ -0,0 +1,147 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.common.schema;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.node.ArrayNode;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+import com.fasterxml.jackson.databind.node.TextNode;
+import java.util.Set;
+import org.apache.pulsar.common.util.ObjectMapperFactory;
+
+/**
+ * Compatibility handling for Avro schema definitions that were written for Avro versions before 1.12.2.
+ *
+ * <p>Up to Avro 1.12.1 the parser accepted a reference to a previously defined named type written as a JSON
+ * object whose {@code type} attribute is the name of the type, for example {@code {"type": "org.example.Color"}},
+ * and resolved it to the named type while ignoring any other attribute of the object. Avro 1.12.2 rejects this
+ * form (AVRO-4176) and only accepts the bare name {@code "org.example.Color"}. Schema definitions in this form
+ * have been accepted and stored by the schema registry, and clients keep producing them, so they have to be
+ * rewritten to the bare name before they are passed to the Avro parser.
+ */
+public final class AvroSchemaCompat {
+
+    /**
+     * The values of {@code type} for which an object defines a schema. For any other value the object is a
+     * reference to a named type.
+     */
+    private static final Set<String> TYPE_KEYWORDS = Set.of(
+            "null", "boolean", "int", "long", "float", "double", "bytes", "string",
+            "record", "error", "enum", "array", "map", "fixed");
+
+    private AvroSchemaCompat() {
+    }
+
+    /**
+     * Rewrite named type references written as {@code {"type": "name"}} objects to the bare {@code "name"}.
+     *
+     * <p>Only positions that hold a schema are visited: the schema itself, field types, array items, map values
+     * and union members. Other JSON values, for example field defaults, are never modified.
+     *
+     * @param schemaDefinition the Avro schema definition as JSON
+     * @return the rewritten definition, or the given string unchanged if it contains no such reference or is not
+     *         valid JSON
+     */
+    public static String normalizeNamedTypeReferences(String schemaDefinition) {
+        if (schemaDefinition == null || schemaDefinition.indexOf('{') < 0) {
+            return schemaDefinition;
+        }
+        JsonNode root;
+        try {
+            root = ObjectMapperFactory.getMapper().reader().readTree(schemaDefinition);
+        } catch (JsonProcessingException e) {
+            // let the Avro parser report the error
+            return schemaDefinition;
+        }
+        NamedTypeReferenceRewriter rewriter = new NamedTypeReferenceRewriter();
+        JsonNode rewritten = rewriter.rewrite(root);
+        if (!rewriter.changed) {
+            return schemaDefinition;
+        }
+        try {
+            return ObjectMapperFactory.getMapper().writer().writeValueAsString(rewritten);
+        } catch (JsonProcessingException e) {
+            return schemaDefinition;
+        }
+    }
+
+    private static final class NamedTypeReferenceRewriter {
+
+        private boolean changed;
+
+        JsonNode rewrite(JsonNode node) {
+            if (node instanceof ArrayNode) {
+                ArrayNode union = (ArrayNode) node;
+                for (int i = 0; i < union.size(); i++) {
+                    union.set(i, rewrite(union.get(i)));
+                }
+                return union;
+            }
+            if (!(node instanceof ObjectNode)) {
+                // a bare name, or something invalid that the Avro parser reports
+                return node;
+            }
+            ObjectNode object = (ObjectNode) node;
+            JsonNode type = object.get("type");
+            if (type == null || !type.isTextual()) {
+                return object;
+            }
+            String typeName = type.textValue();
+            if (!TYPE_KEYWORDS.contains(typeName)) {
+                changed = true;
+                return TextNode.valueOf(typeName);
+            }
+            switch (typeName) {
+                case "record":
+                case "error":
+                    rewriteFieldTypes(object.get("fields"));
+                    break;
+                case "array":
+                    rewriteChild(object, "items");
+                    break;
+                case "map":
+                    rewriteChild(object, "values");
+                    break;
+                default:
+                    // primitive, enum and fixed types have no nested schema
+                    break;
+            }
+            return object;
+        }
+
+        private void rewriteFieldTypes(JsonNode fields) {
+            if (fields == null || !fields.isArray()) {
+                return;
+            }
+            for (JsonNode field : fields) {
+                if (field instanceof ObjectNode) {
+                    rewriteChild((ObjectNode) field, "type");
+                }
+            }
+        }
+
+        private void rewriteChild(ObjectNode object, String name) {
+            JsonNode child = object.get(name);
+            if (child != null) {
+                object.set(name, rewrite(child));
+            }
+        }
+    }
+}