[fix][broker] Use compatible Avro name validator in JsonSchemaCompatibilityCheck (#25255)

Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>

Author: 2 people

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;

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;