eolivelli commented on a change in pull request #9612:
URL: https://github.com/apache/pulsar/pull/9612#discussion_r578352086
##########
File path:
pulsar-broker/src/test/java/org/apache/pulsar/schema/compatibility/SchemaCompatibilityCheckTest.java
##########
@@ -294,6 +298,53 @@ public void
testIsAutoUpdateSchema(SchemaCompatibilityStrategy schemaCompatibili
producer.close();
}
+ @Test
+ public void testSchemaComparison() throws Exception {
+ final String tenant = PUBLIC_TENANT;
+ final String topic = "test-schema-comparison";
+
+ String namespace = "test-namespace-" + randomName(16);
+ String fqtn = TopicName.get(
+ TopicDomain.persistent.value(),
+ tenant,
+ namespace,
+ topic
+ ).toString();
+
+ NamespaceName namespaceName = NamespaceName.get(tenant, namespace);
+
+ admin.namespaces().createNamespace(
+ tenant + "/" + namespace,
+ Sets.newHashSet(CLUSTER_NAME)
+ );
+
+
assertEquals(admin.namespaces().getSchemaCompatibilityStrategy(namespaceName.toString()),
+ SchemaCompatibilityStrategy.FULL);
+ byte[] changeSchemaBytes = (new
String(Schema.AVRO(Schemas.PersonOne.class)
+ .getSchemaInfo().getSchema(), UTF_8) + "/n /n
/n").getBytes();
+ SchemaInfo schemaInfo =
SchemaInfo.builder().type(SchemaType.AVRO).schema(changeSchemaBytes).build();
+ admin.schemas().createSchema(fqtn, schemaInfo);
+
+
admin.namespaces().setIsAllowAutoUpdateSchema(namespaceName.toString(), false);
+ ProducerBuilder<Schemas.PersonOne> producerOneBuilder = pulsarClient
+ .newProducer(Schema.AVRO(Schemas.PersonOne.class))
+ .topic(fqtn);
+ producerOneBuilder.create().close();
+
+ assertArrayEquals(changeSchemaBytes,
admin.schemas().getSchemaInfo(fqtn).getSchema());
+
+ ProducerBuilder<Schemas.PersonThree> producerThreeBuilder =
pulsarClient
+ .newProducer(Schema.AVRO(Schemas.PersonThree.class))
+ .topic(fqtn);
+
+ try {
+ producerThreeBuilder.create();
+ } catch (Exception e) {
+ e.printStackTrace();
Review comment:
please use logger
##########
File path:
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/SchemaRegistryServiceImpl.java
##########
@@ -293,12 +295,28 @@ private void checkCompatible(SchemaAndMetadata
existingSchema, SchemaData newSch
SchemaData schemaData) {
final CompletableFuture<SchemaVersion> completableFuture = new
CompletableFuture<>();
SchemaVersion schemaVersion;
- for (SchemaAndMetadata schemaAndMetadata : schemaAndMetadataList) {
- if
(Arrays.equals(hashFunction.hashBytes(schemaAndMetadata.schema.getData()).asBytes(),
- hashFunction.hashBytes(schemaData.getData()).asBytes())) {
- schemaVersion = schemaAndMetadata.version;
- completableFuture.complete(schemaVersion);
- return completableFuture;
+ if (schemaData.getData().length != 0) {
+ Schema.Parser parser = new Schema.Parser();
+ Schema newSchema = parser.parse(new String(schemaData.getData(),
UTF_8));
Review comment:
are we sure that here we are dealing with an Avro schema ?
what happens in case of a non-Avro schema ?
##########
File path:
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/SchemaRegistryServiceImpl.java
##########
@@ -293,12 +295,28 @@ private void checkCompatible(SchemaAndMetadata
existingSchema, SchemaData newSch
SchemaData schemaData) {
final CompletableFuture<SchemaVersion> completableFuture = new
CompletableFuture<>();
SchemaVersion schemaVersion;
- for (SchemaAndMetadata schemaAndMetadata : schemaAndMetadataList) {
- if
(Arrays.equals(hashFunction.hashBytes(schemaAndMetadata.schema.getData()).asBytes(),
- hashFunction.hashBytes(schemaData.getData()).asBytes())) {
- schemaVersion = schemaAndMetadata.version;
- completableFuture.complete(schemaVersion);
- return completableFuture;
+ if (schemaData.getData().length != 0) {
Review comment:
you added one test, which one of the two branches is covered by the new
test ?
Can you add a test for the other branch or point to an existing test ?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
