[fix][admin] Fix KeyValue schema compatibility check caused OOM (#21645)

Author: Technoboy-

pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/SchemasResourceBase.java

@@ -146,8 +146,13 @@ public CompletableFuture<Pair<Boolean, SchemaCompatibilityStrategy>> testCompati
                 .thenCompose(__ -> getSchemaCompatibilityStrategyAsync())
                 .thenCompose(strategy -> {
                     String schemaId = getSchemaId();
+                    final SchemaType schemaType = SchemaType.valueOf(payload.getType());
+                    byte[] data = payload.getSchema().getBytes(StandardCharsets.UTF_8);
+                    if (schemaType.getValue() == SchemaType.KEY_VALUE.getValue()) {
+                        data = SchemaUtils.convertKeyValueDataStringToSchemaInfoSchema(data);
+                    }
                     return pulsar().getSchemaRegistryService().isCompatible(schemaId,
-                            SchemaData.builder().data(payload.getSchema().getBytes(StandardCharsets.UTF_8))
+                            SchemaData.builder().data(data)
                                     .isDeleted(false)
                                     .timestamp(clock.millis()).type(SchemaType.valueOf(payload.getType()))
                                     .user(defaultIfEmpty(clientAppId(), ""))

pulsar-broker/src/test/java/org/apache/pulsar/broker/service/schema/SchemaServiceTest.java

@@ -18,6 +18,7 @@
  */
 package org.apache.pulsar.broker.service.schema;
 
+import static org.testng.Assert.assertThrows;
 import static org.testng.Assert.fail;
 import static org.testng.AssertJUnit.assertEquals;
 import static org.testng.AssertJUnit.assertFalse;
@@ -40,16 +41,23 @@
 import java.util.UUID;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ExecutionException;
+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.broker.stats.PrometheusMetricsTest;
 import org.apache.pulsar.broker.stats.prometheus.PrometheusMetricsGenerator;
+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;
+import org.apache.pulsar.common.protocol.schema.IsCompatibilityResponse;
 import org.apache.pulsar.common.protocol.schema.SchemaData;
 import org.apache.pulsar.common.protocol.schema.SchemaVersion;
+import org.apache.pulsar.common.schema.KeyValueEncodingType;
 import org.apache.pulsar.common.schema.LongSchemaVersion;
+import org.apache.pulsar.common.schema.SchemaInfo;
+import org.apache.pulsar.common.schema.SchemaInfoWithVersion;
 import org.apache.pulsar.common.schema.SchemaType;
 import org.testng.Assert;
 import org.testng.annotations.AfterMethod;
@@ -385,4 +393,32 @@ private static SchemaData getSchemaData(String schemaJson) {
     private SchemaVersion version(long version) {
         return new LongSchemaVersion(version);
     }
+
+    @Test
+    public void testKeyValueSchema() throws Exception {
+        final String topicName = "persistent://public/default/testKeyValueSchema";
+        admin.topics().createNonPartitionedTopic(BrokerTestUtil.newUniqueName(topicName));
+
+        final SchemaInfo schemaInfo = KeyValueSchemaInfo.encodeKeyValueSchemaInfo(
+                "keyValue",
+                SchemaInfo.builder().type(SchemaType.STRING).schema(new byte[0])
+                        .build(),
+                SchemaInfo.builder().type(SchemaType.BOOLEAN).schema(new byte[0])
+                        .build(), KeyValueEncodingType.SEPARATED);
+        assertThrows(PulsarAdminException.ServerSideErrorException.class, () -> admin.schemas().testCompatibility(topicName, schemaInfo));
+        admin.schemas().createSchema(topicName, schemaInfo);
+
+        final IsCompatibilityResponse isCompatibilityResponse = admin.schemas().testCompatibility(topicName, schemaInfo);
+        Assert.assertTrue(isCompatibilityResponse.isCompatibility());
+
+        final SchemaInfoWithVersion schemaInfoWithVersion = admin.schemas().getSchemaInfoWithVersion(topicName);
+        Assert.assertEquals(schemaInfoWithVersion.getVersion(), 0);
+
+        final Long version1 = admin.schemas().getVersionBySchema(topicName, schemaInfo);
+        Assert.assertEquals(version1, 0);
+
+        final Long version2 = admin.schemas().getVersionBySchema(topicName, schemaInfoWithVersion.getSchemaInfo());
+        Assert.assertEquals(version2, 0);
+
+    }
 }