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); + } +}