[GitHub] [pulsar] eolivelli commented on a change in pull request #9612: [Schema] Schema comparison logic change.

[GitBox <gi...@apache.org>]

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:
users@infra.apache.org