This is an automated email from the ASF dual-hosted git repository.
baodi pushed a commit to branch branch-2.10
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/branch-2.10 by this push:
new 573bf6d3f6e [fix][schema] Error checking schema compatibility on a
schema-less topic via REST API (#22720)
573bf6d3f6e is described below
commit 573bf6d3f6ecfec95ce39307f664c43512111471
Author: Baodi Shi <[email protected]>
AuthorDate: Thu May 16 18:02:17 2024 +0800
[fix][schema] Error checking schema compatibility on a schema-less topic
via REST API (#22720)
(cherry picked from commit 101aee4)
---
.../schema/AvroSchemaBasedCompatibilityCheck.java | 6 ++--
.../ProtobufNativeSchemaCompatibilityCheck.java | 4 ++-
.../service/schema/SchemaRegistryServiceImpl.java | 2 +-
.../exceptions/IncompatibleSchemaException.java | 4 +++
.../pulsar/broker/admin/AdminApiSchemaTest.java | 32 ++++++++++++++++++++++
5 files changed, 44 insertions(+), 4 deletions(-)
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 ce4bb2a4968..1f1635e9347 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
@@ -64,8 +64,10 @@ abstract class AvroSchemaBasedCompatibilityCheck implements
SchemaCompatibilityC
log.warn("Error during schema parsing: {}", e.getMessage());
throw new IncompatibleSchemaException(e);
} catch (SchemaValidationException e) {
- log.warn("Error during schema compatibility check: {}",
e.getMessage());
- throw new IncompatibleSchemaException(e);
+ String msg = String.format("Error during schema compatibility
check with strategy %s: %s: %s",
+ strategy, e.getClass().getName(), e.getMessage());
+ log.warn(msg);
+ throw new IncompatibleSchemaException(msg, e);
}
}
diff --git
a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/ProtobufNativeSchemaCompatibilityCheck.java
b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/ProtobufNativeSchemaCompatibilityCheck.java
index 58545370565..da3dd8138fc 100644
---
a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/ProtobufNativeSchemaCompatibilityCheck.java
+++
b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/ProtobufNativeSchemaCompatibilityCheck.java
@@ -67,7 +67,9 @@ public class ProtobufNativeSchemaCompatibilityCheck
implements SchemaCompatibili
private void checkRootMessageChange(Descriptor fromDescriptor, Descriptor
toDescriptor,
SchemaCompatibilityStrategy
strategy) throws IncompatibleSchemaException {
if (!fromDescriptor.getFullName().equals(toDescriptor.getFullName())) {
- throw new IncompatibleSchemaException("Protobuf root message isn't
allow change!");
+ throw new IncompatibleSchemaException("Protobuf root message
change is not allowed under the '"
+ + strategy + "' strategy. Original message name: '" +
fromDescriptor.getFullName()
+ + "', new message name: '" + toDescriptor.getFullName() +
"'.");
}
}
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 9cfcce419fd..5f1c67e7bdc 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
@@ -394,7 +394,7 @@ public class SchemaRegistryServiceImpl implements
SchemaRegistryService {
}
return result;
} else {
- return FutureUtils.exception(new
IncompatibleSchemaException("Do not have existing schema."));
+ return CompletableFuture.completedFuture(null);
}
});
}
diff --git
a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/exceptions/IncompatibleSchemaException.java
b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/exceptions/IncompatibleSchemaException.java
index 759cc08e3ff..a4731161b29 100644
---
a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/exceptions/IncompatibleSchemaException.java
+++
b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/exceptions/IncompatibleSchemaException.java
@@ -33,6 +33,10 @@ public class IncompatibleSchemaException extends
SchemaException {
super(message);
}
+ public IncompatibleSchemaException(String message, Throwable e) {
+ super(message, e);
+ }
+
public IncompatibleSchemaException(Throwable e) {
super(e);
}
diff --git
a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiSchemaTest.java
b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiSchemaTest.java
index 6d5f0e5b489..4d545d141eb 100644
---
a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiSchemaTest.java
+++
b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiSchemaTest.java
@@ -49,6 +49,8 @@ import
org.apache.pulsar.common.policies.data.PersistentTopicInternalStats;
import
org.apache.pulsar.common.policies.data.SchemaAutoUpdateCompatibilityStrategy;
import org.apache.pulsar.common.policies.data.SchemaCompatibilityStrategy;
import org.apache.pulsar.common.policies.data.TenantInfoImpl;
+import org.apache.pulsar.common.protocol.schema.IsCompatibilityResponse;
+import org.apache.pulsar.common.protocol.schema.PostSchemaPayload;
import org.apache.pulsar.common.schema.SchemaInfo;
import org.apache.pulsar.common.schema.SchemaInfoWithVersion;
import org.awaitility.Awaitility;
@@ -415,4 +417,34 @@ public class AdminApiSchemaTest extends
MockedPulsarServiceBaseTest {
admin.namespaces().getSchemaCompatibilityStrategy(schemaCompatibilityNamespace),
SchemaCompatibilityStrategy.UNDEFINED));
}
+
+ @Test
+ public void testCompatibilityWithEmpty() throws Exception {
+ List<Schema<?>> checkSchemas = List.of(
+ Schema.STRING,
+
Schema.JSON(SchemaDefinition.builder().withPojo(Foo.class).withProperties(PROPS).build()),
+
Schema.AVRO(SchemaDefinition.builder().withPojo(Foo.class).withProperties(PROPS).build()),
+ Schema.KeyValue(Schema.STRING, Schema.STRING)
+ );
+ for (Schema<?> schema : checkSchemas) {
+ SchemaInfo schemaInfo = schema.getSchemaInfo();
+ String topicName = schemaCompatibilityNamespace +
"/testCompatibilityWithEmpty";
+ PostSchemaPayload postSchemaPayload = new
PostSchemaPayload(schemaInfo.getType().toString(),
+ schemaInfo.getSchemaDefinition(), new HashMap<>());
+
+ // check compatibility with empty schema
+ IsCompatibilityResponse isCompatibilityResponse =
+ admin.schemas().testCompatibility(topicName,
postSchemaPayload);
+ assertTrue(isCompatibilityResponse.isCompatibility());
+
assertEquals(isCompatibilityResponse.getSchemaCompatibilityStrategy(),
SchemaCompatibilityStrategy.FULL.name());
+
+ // set schema compatibility strategy is FULL_TRANSITIVE to cover
checkCompatibilityWithAll
+
admin.namespaces().setSchemaCompatibilityStrategy(schemaCompatibilityNamespace,
SchemaCompatibilityStrategy.FULL_TRANSITIVE);
+ isCompatibilityResponse =
admin.schemas().testCompatibility(topicName, postSchemaPayload);
+ assertTrue(isCompatibilityResponse.isCompatibility());
+
assertEquals(isCompatibilityResponse.getSchemaCompatibilityStrategy(),
SchemaCompatibilityStrategy.FULL_TRANSITIVE.name());
+ // set back to FULL
+
admin.namespaces().setSchemaCompatibilityStrategy(schemaCompatibilityNamespace,
SchemaCompatibilityStrategy.FULL);
+ }
+ }
}
