[fix][schema] Error checking schema compatibility on a schema-less to…

…pic via REST API (apache#22720) (cherry picked from commit 101aee4)
ascentstream · Jun 7, 2024 · 4a2fb34 · 4a2fb34
1 parent 9c345f1
commit 4a2fb34
Show file tree

Hide file tree

Showing 5 changed files with 44 additions and 4 deletions.
diff --git a/.../main/java/org/apache/pulsar/broker/service/schema/AvroSchemaBasedCompatibilityCheck.java b/.../main/java/org/apache/pulsar/broker/service/schema/AvroSchemaBasedCompatibilityCheck.java
@@ -64,8 +64,10 @@ public void checkCompatible(Iterable<SchemaData> from, SchemaData to, SchemaComp
             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/.../java/org/apache/pulsar/broker/service/schema/ProtobufNativeSchemaCompatibilityCheck.java b/.../java/org/apache/pulsar/broker/service/schema/ProtobufNativeSchemaCompatibilityCheck.java
@@ -67,7 +67,9 @@ public void checkCompatible(Iterable<SchemaData> from, SchemaData to, SchemaComp
     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/...oker/src/main/java/org/apache/pulsar/broker/service/schema/SchemaRegistryServiceImpl.java b/...oker/src/main/java/org/apache/pulsar/broker/service/schema/SchemaRegistryServiceImpl.java
@@ -394,7 +394,7 @@ private CompletableFuture<Void> checkCompatibilityWithLatest(String schemaId, Sc
                 }
                 return result;
             } else {
-                return FutureUtils.exception(new IncompatibleSchemaException("Do not have existing schema."));
+                return CompletableFuture.completedFuture(null);
             }
         });
     }

diff --git a/.../java/org/apache/pulsar/broker/service/schema/exceptions/IncompatibleSchemaException.java b/.../java/org/apache/pulsar/broker/service/schema/exceptions/IncompatibleSchemaException.java
@@ -33,6 +33,10 @@ public IncompatibleSchemaException(String message) {
         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
@@ -49,6 +49,8 @@
 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 void testGetSchemaCompatibilityStrategyWhenSetBrokerLevelAndSchemaAutoUpd
                 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);
+        }
+    }
 }