Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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;
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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;
Expand Down
Loading