This is an automated email from the ASF dual-hosted git repository. baodi pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/master by this push: new 101aee4543f [fix][schema] Error checking schema compatibility on a schema-less topic via REST API (#22720) 101aee4543f is described below commit 101aee4543fb66035165d8744def630f9a9c3a59 Author: Baodi Shi <ba...@apache.org> AuthorDate: Thu May 16 17:36:57 2024 +0800 [fix][schema] Error checking schema compatibility on a schema-less topic via REST API (#22720) --- .../schema/AvroSchemaBasedCompatibilityCheck.java | 6 +++-- .../ProtobufNativeSchemaCompatibilityCheck.java | 4 ++- .../service/schema/SchemaRegistryServiceImpl.java | 2 +- .../exceptions/IncompatibleSchemaException.java | 4 +++ .../pulsar/broker/admin/AdminApiSchemaTest.java | 30 ++++++++++++++++++++++ .../broker/service/schema/SchemaServiceTest.java | 4 +-- 6 files changed, 43 insertions(+), 7 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 1e75834a129..e5fc7800c51 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 16b3b33ec78..fc935e80dca 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 ae56df248d8..903f57cb780 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 @@ -473,7 +473,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 c1a2d9fd703..bbe2f4111d7 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 f67bd6fcfce..34d7dbeb818 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 @@ -467,4 +467,34 @@ public class AdminApiSchemaTest extends MockedPulsarServiceBaseTest { assertTrue(e.getMessage().contains("Incompatible schema: exists schema type STRING, new schema type INT8")); } } + + @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); + } + } } diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/schema/SchemaServiceTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/schema/SchemaServiceTest.java index 3a4016eb79c..fbf8c5cc154 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/schema/SchemaServiceTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/schema/SchemaServiceTest.java @@ -20,7 +20,6 @@ package org.apache.pulsar.broker.service.schema; import static org.apache.pulsar.broker.stats.prometheus.PrometheusMetricsClient.Metric; import static org.apache.pulsar.broker.stats.prometheus.PrometheusMetricsClient.parseMetrics; -import static org.testng.Assert.assertThrows; import static org.testng.Assert.fail; import static org.testng.AssertJUnit.assertEquals; import static org.testng.AssertJUnit.assertFalse; @@ -48,7 +47,6 @@ import org.apache.pulsar.broker.BrokerTestUtil; import org.apache.pulsar.broker.PulsarServerException; import org.apache.pulsar.broker.auth.MockedPulsarServiceBaseTest; import org.apache.pulsar.broker.service.schema.SchemaRegistry.SchemaAndMetadata; -import org.apache.pulsar.client.admin.PulsarAdminException; import org.apache.pulsar.client.impl.schema.KeyValueSchemaInfo; import org.apache.pulsar.common.naming.TopicName; import org.apache.pulsar.common.policies.data.SchemaCompatibilityStrategy; @@ -407,7 +405,7 @@ public class SchemaServiceTest extends MockedPulsarServiceBaseTest { .build(), SchemaInfo.builder().type(SchemaType.BOOLEAN).schema(new byte[0]) .build(), KeyValueEncodingType.SEPARATED); - assertThrows(PulsarAdminException.ServerSideErrorException.class, () -> admin.schemas().testCompatibility(topicName, schemaInfo)); + Assert.assertTrue(admin.schemas().testCompatibility(topicName, schemaInfo).isCompatibility()); admin.schemas().createSchema(topicName, schemaInfo); final IsCompatibilityResponse isCompatibilityResponse = admin.schemas().testCompatibility(topicName, schemaInfo);