(pulsar) branch branch-4.1 updated: [fix][broker] Use compatible Avro name validator in JsonSchemaCompatibilityCheck (#25255)

[te...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

technoboy pushed a commit to branch branch-4.1
in repository https://gitbox.apache.org/repos/asf/pulsar.git


The following commit(s) were added to refs/heads/branch-4.1 by this push:
     new 55a9fdbeaa0 [fix][broker] Use compatible Avro name validator in JsonSchemaCompatibilityCheck (#25255)
55a9fdbeaa0 is described below

commit 55a9fdbeaa0b42a5006f337db6863bc85b71d2f1
Author: Qiang Zhao <ma...@apache.org>
AuthorDate: Wed Feb 25 14:03:49 2026 +0800

    [fix][broker] Use compatible Avro name validator in JsonSchemaCompatibilityCheck (#25255)
    
    Co-authored-by: Claude Opus 4.6 <no...@anthropic.com>
---
 .../schema/JsonSchemaCompatibilityCheck.java       |  3 ++-
 .../schema/JsonSchemaCompatibilityCheckTest.java   | 28 ++++++++++++++++++++++
 2 files changed, 30 insertions(+), 1 deletion(-)

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 cbdca3c4732..94afdd14620 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 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;
 import org.apache.pulsar.common.protocol.schema.SchemaData;
 import org.apache.pulsar.common.schema.SchemaType;
@@ -91,7 +92,7 @@ public class JsonSchemaCompatibilityCheck extends AvroSchemaBasedCompatibilityCh
     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 b68ecb08799..b192110c2d7 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 class JsonSchemaCompatibilityCheckTest extends BaseAvroSchemaCompatibilit
         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;