From ce3a46abaadcd07f9bc256bb5a404a85c1b3a775 Mon Sep 17 00:00:00 2001 From: Lari Hotari Date: Tue, 15 Sep 2026 12:11:40 +0300 Subject: [PATCH] [fix][broker] Keep accepting Avro named type references written as objects Avro 1.12.2 changed the schema parser so that a reference to a named type must be written as the bare name (AVRO-4176). Up to Avro 1.12.1 the reference could also be written as an object whose "type" is the name, for example {"type": "org.example.shapes.Color"}, and the parser resolved it to the named type while ignoring any other attribute. After the Avro upgrade such schema definitions fail with SchemaParseException: A schema "type" MUST be a primitive type or one of "enum", "fixed", "record", "error", "array" or "map". They were accepted and stored by the schema registry before, and clients may still produce them, so the broker fails to validate, look up and check compatibility of these schemas and clients fail to consume from topics whose stored schema uses this form. Rewrite such objects to the bare name before a schema definition is handed to the Avro parser. The rewrite in pulsar-common follows the schema structure, so field defaults and other JSON values are never modified, and returns the input unchanged when there is nothing to rewrite. The broker parses schema definitions through a single helper in StructSchemaDataValidator, which also applies the compatible name validator, and the client applies the rewrite in SchemaUtil and GenericJsonRecord. Stored schema data is not modified. --- .../AvroSchemaBasedCompatibilityCheck.java | 10 +- .../schema/JsonSchemaCompatibilityCheck.java | 6 +- .../schema/SchemaRegistryServiceImpl.java | 9 +- .../validator/StructSchemaDataValidator.java | 19 ++- .../BaseAvroSchemaCompatibilityTest.java | 20 +++ .../validator/SchemaDataValidatorTest.java | 33 ++++ .../schema/generic/GenericJsonRecord.java | 3 +- .../client/impl/schema/util/SchemaUtil.java | 3 +- .../impl/schema/util/SchemaUtilTest.java | 19 +++ .../common/schema/AvroSchemaCompat.java | 147 ++++++++++++++++++ .../common/schema/AvroSchemaCompatTest.java | 101 ++++++++++++ 11 files changed, 348 insertions(+), 22 deletions(-) create mode 100644 pulsar-common/src/main/java/org/apache/pulsar/common/schema/AvroSchemaCompat.java create mode 100644 pulsar-common/src/test/java/org/apache/pulsar/common/schema/AvroSchemaCompatTest.java diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/AvroSchemaBasedCompatibilityCheck.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/AvroSchemaBasedCompatibilityCheck.java index 666fd28f69e7a..d0a625daea175 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/AvroSchemaBasedCompatibilityCheck.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/AvroSchemaBasedCompatibilityCheck.java @@ -52,14 +52,10 @@ public void checkCompatible(Iterable 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) { diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/JsonSchemaCompatibilityCheck.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/JsonSchemaCompatibilityCheck.java index dc8be3e0651a7..4811277d8a301 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/JsonSchemaCompatibilityCheck.java +++ b/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; diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/SchemaRegistryServiceImpl.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/SchemaRegistryServiceImpl.java index b47c6d29f3776..592bc3a1c51e6 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/SchemaRegistryServiceImpl.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/SchemaRegistryServiceImpl.java @@ -403,14 +403,13 @@ public CompletableFuture getSchemaVersionBySchemaData( final CompletableFuture 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); diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/validator/StructSchemaDataValidator.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/validator/StructSchemaDataValidator.java index 267e42c0ca3c8..84c0cdac33eef 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/validator/StructSchemaDataValidator.java +++ b/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: { diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/schema/BaseAvroSchemaCompatibilityTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/schema/BaseAvroSchemaCompatibilityTest.java index cbb59c2ecde9d..5fff51cbcd7fe 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/schema/BaseAvroSchemaCompatibilityTest.java +++ b/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(); diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/schema/validator/SchemaDataValidatorTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/schema/validator/SchemaDataValidatorTest.java index b001d384bb95c..e246fba977cbf 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/schema/validator/SchemaDataValidatorTest.java +++ b/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"); + } + } diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/generic/GenericJsonRecord.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/generic/GenericJsonRecord.java index 69010b68c401a..c0d83c4563dc9 100644 --- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/generic/GenericJsonRecord.java +++ b/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 diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/util/SchemaUtil.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/util/SchemaUtil.java index 5fdffb5634926..0be32b5ad49f8 100644 --- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/util/SchemaUtil.java +++ b/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 SchemaInfo parseSchemaInfo(SchemaDefinition schemaDefinition, SchemaType schemaType) { diff --git a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/schema/util/SchemaUtilTest.java b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/schema/util/SchemaUtilTest.java index d80a25873a8ee..c8f7a0f98a170 100644 --- a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/schema/util/SchemaUtilTest.java +++ b/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"); + } } diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/schema/AvroSchemaCompat.java b/pulsar-common/src/main/java/org/apache/pulsar/common/schema/AvroSchemaCompat.java new file mode 100644 index 0000000000000..9e0dfa05fe75c --- /dev/null +++ b/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. + * + *

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 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"}. + * + *

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)); + } + } + } +} diff --git a/pulsar-common/src/test/java/org/apache/pulsar/common/schema/AvroSchemaCompatTest.java b/pulsar-common/src/test/java/org/apache/pulsar/common/schema/AvroSchemaCompatTest.java new file mode 100644 index 0000000000000..d64745ab1541d --- /dev/null +++ b/pulsar-common/src/test/java/org/apache/pulsar/common/schema/AvroSchemaCompatTest.java @@ -0,0 +1,101 @@ +/* + * 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 static org.testng.Assert.assertEquals; +import static org.testng.Assert.assertSame; +import com.fasterxml.jackson.databind.JsonNode; +import org.apache.pulsar.common.util.ObjectMapperFactory; +import org.testng.annotations.Test; + +public class AvroSchemaCompatTest { + + private static final String COLOR_ENUM = "{\"type\":\"enum\",\"name\":\"Color\"," + + "\"namespace\":\"org.example.shapes\",\"symbols\":[\"RED\",\"BLUE\"]}"; + + @Test + public void testNamedTypeReferencesAreRewrittenInAllSchemaPositions() throws Exception { + String legacy = "{\"type\":\"record\",\"name\":\"Drawing\",\"namespace\":\"org.example.shapes\",\"fields\":[" + + "{\"name\":\"background\",\"type\":" + COLOR_ENUM + "}," + + "{\"name\":\"outline\",\"type\":{\"type\":\"org.example.shapes.Color\"}}," + + "{\"name\":\"palette\",\"type\":{\"type\":\"array\"," + + "\"items\":{\"type\":\"org.example.shapes.Color\"}}}," + + "{\"name\":\"labels\",\"type\":{\"type\":\"map\"," + + "\"values\":{\"type\":\"org.example.shapes.Color\"}}}," + + "{\"name\":\"highlight\",\"type\":[\"null\",{\"type\":\"org.example.shapes.Color\"}]," + + "\"default\":null}" + + "]}"; + String expected = legacy.replace("{\"type\":\"org.example.shapes.Color\"}", "\"org.example.shapes.Color\""); + + assertJsonEquals(AvroSchemaCompat.normalizeNamedTypeReferences(legacy), expected); + } + + @Test + public void testNestedRecordsAreVisited() throws Exception { + String legacy = "{\"type\":\"record\",\"name\":\"Canvas\",\"namespace\":\"org.example.shapes\",\"fields\":[" + + "{\"name\":\"colors\",\"type\":{\"type\":\"array\",\"items\":" + COLOR_ENUM + "}}," + + "{\"name\":\"frame\",\"type\":{\"type\":\"record\",\"name\":\"Frame\",\"fields\":[" + + "{\"name\":\"color\",\"type\":{\"type\":\"org.example.shapes.Color\"}}]}}" + + "]}"; + String expected = legacy.replace("{\"type\":\"org.example.shapes.Color\"}", "\"org.example.shapes.Color\""); + + assertJsonEquals(AvroSchemaCompat.normalizeNamedTypeReferences(legacy), expected); + } + + @Test + public void testTopLevelNamedTypeReferenceIsRewritten() { + assertEquals(AvroSchemaCompat.normalizeNamedTypeReferences("{\"type\":\"org.example.shapes.Color\"}"), + "\"org.example.shapes.Color\""); + } + + @Test + public void testDefaultValuesAreNotModified() throws Exception { + // the default of "metadata" is a record value that happens to have a field named "type" + String schema = "{\"type\":\"record\",\"name\":\"Document\",\"fields\":[" + + "{\"name\":\"metadata\",\"type\":{\"type\":\"record\",\"name\":\"Metadata\",\"fields\":[" + + "{\"name\":\"type\",\"type\":\"string\"}]},\"default\":{\"type\":\"org.example.NotASchema\"}}," + + "{\"name\":\"previous\",\"type\":{\"type\":\"Metadata\"},\"default\":{\"type\":\"draft\"}}" + + "]}"; + String expected = schema.replace("{\"type\":\"Metadata\"}", "\"Metadata\""); + + assertJsonEquals(AvroSchemaCompat.normalizeNamedTypeReferences(schema), expected); + } + + @Test + public void testSchemaWithoutLegacyReferencesIsReturnedUnchanged() { + String schema = "{\"type\":\"record\",\"name\":\"Point\",\"fields\":[" + + "{\"name\":\"x\",\"type\":\"double\"},{\"name\":\"y\",\"type\":{\"type\":\"double\"}}," + + "{\"name\":\"tags\",\"type\":{\"type\":\"array\",\"items\":\"string\"}}]}"; + assertSame(AvroSchemaCompat.normalizeNamedTypeReferences(schema), schema); + assertSame(AvroSchemaCompat.normalizeNamedTypeReferences("\"string\""), "\"string\""); + } + + @Test + public void testInvalidJsonIsReturnedUnchanged() { + String notJson = "{\"type\":\"record\","; + assertSame(AvroSchemaCompat.normalizeNamedTypeReferences(notJson), notJson); + assertSame(AvroSchemaCompat.normalizeNamedTypeReferences(null), null); + } + + private static void assertJsonEquals(String actual, String expected) throws Exception { + JsonNode actualNode = ObjectMapperFactory.getMapper().reader().readTree(actual); + JsonNode expectedNode = ObjectMapperFactory.getMapper().reader().readTree(expected); + assertEquals(actualNode, expectedNode); + } +}