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 cbdca3c47322c..94afdd14620bf 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 @@ -24,6 +24,7 @@ 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; import org.apache.pulsar.common.protocol.schema.SchemaData; import org.apache.pulsar.common.schema.SchemaType; @@ -91,7 +92,7 @@ private void isCompatibleJsonSchema(SchemaData from, SchemaData to) throws Incom private boolean isAvroSchema(SchemaData schemaData) { try { - Schema.Parser fromParser = new Schema.Parser(); + Schema.Parser fromParser = new Schema.Parser(StructSchemaDataValidator.COMPATIBLE_NAME_VALIDATOR); fromParser.setValidateDefaults(false); Schema fromSchema = fromParser.parse(new String(schemaData.getData(), UTF_8)); return true; diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/schema/JsonSchemaCompatibilityCheckTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/schema/JsonSchemaCompatibilityCheckTest.java index b68ecb08799ad..b192110c2d7bf 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/schema/JsonSchemaCompatibilityCheckTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/schema/JsonSchemaCompatibilityCheckTest.java @@ -18,6 +18,7 @@ */ package org.apache.pulsar.broker.service.schema; +import static java.nio.charset.StandardCharsets.UTF_8; import com.fasterxml.jackson.core.JsonProcessingException; import com.fasterxml.jackson.databind.ObjectMapper; import com.fasterxml.jackson.module.jsonSchema.JsonSchema; @@ -60,6 +61,33 @@ public void testJsonSchemaBackwardsCompatibility() throws JsonProcessingExceptio Assert.assertTrue(jsonSchemaCompatibilityCheck.isCompatible(from, to, SchemaCompatibilityStrategy.FULL)); } + @Test + public void testSchemaWithDollarSignInRecordNameRejectsIncompatibleChange() { + // Schema v1: has field1 (string) + String schemaV1 = + "{\"type\":\"record\",\"name\":\"Outer$Inner\",\"namespace\":\"org.example\"," + + "\"fields\":[{\"name\":\"field1\",\"type\":\"string\"}]}"; + // Schema v2: removed field1, added field2 without default — NOT backward compatible + String schemaV2 = + "{\"type\":\"record\",\"name\":\"Outer$Inner\",\"namespace\":\"org.example\"," + + "\"fields\":[{\"name\":\"field2\",\"type\":\"string\"}]}"; + SchemaData from = SchemaData.builder() + .data(schemaV1.getBytes(UTF_8)) + .type(SchemaType.JSON) + .build(); + SchemaData to = SchemaData.builder() + .data(schemaV2.getBytes(UTF_8)) + .type(SchemaType.JSON) + .build(); + JsonSchemaCompatibilityCheck check = new JsonSchemaCompatibilityCheck(); + // Without the fix, isAvroSchema() rejects '$' and the compatibility check is + // skipped entirely (falls through to "corrupted, allow overwrite"), so this + // would incorrectly return true. + // With the fix, isAvroSchema() recognizes these as valid Avro schemas and the + // Avro compatibility check correctly detects the incompatibility. + Assert.assertFalse(check.isCompatible(from, to, SchemaCompatibilityStrategy.BACKWARD)); + } + @Data private static class Foo { private String field1;