Support multiple Avro schema version in Pulsar SQL (#4847)

Support multiple Avro schema version in Pulsar SQL

Author: congbobo184

pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/SchemasResource.java

@@ -54,7 +54,7 @@
 
 import org.apache.pulsar.broker.admin.AdminResource;
 import org.apache.pulsar.broker.service.schema.exceptions.IncompatibleSchemaException;
-import org.apache.pulsar.broker.service.schema.LongSchemaVersion;
+import org.apache.pulsar.common.schema.LongSchemaVersion;
 import org.apache.pulsar.common.policies.data.Policies;
 import org.apache.pulsar.common.policies.data.SchemaCompatibilityStrategy;
 import org.apache.pulsar.broker.service.schema.SchemaRegistry.SchemaAndMetadata;

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/BookkeeperSchemaStorage.java

@@ -52,6 +52,7 @@
 import org.apache.pulsar.broker.ServiceConfiguration;
 import org.apache.pulsar.common.protocol.schema.SchemaStorage;
 import org.apache.pulsar.common.protocol.schema.SchemaVersion;
+import org.apache.pulsar.common.schema.LongSchemaVersion;
 import org.apache.pulsar.common.protocol.schema.StoredSchema;
 import org.apache.pulsar.zookeeper.ZooKeeperCache;
 import org.apache.zookeeper.CreateMode;

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/SchemaRegistryServiceImpl.java

@@ -50,6 +50,7 @@
 import org.apache.pulsar.common.policies.data.SchemaCompatibilityStrategy;
 import org.apache.pulsar.common.protocol.schema.SchemaData;
 import org.apache.pulsar.common.protocol.schema.SchemaHash;
+import org.apache.pulsar.common.schema.LongSchemaVersion;
 import org.apache.pulsar.common.protocol.schema.SchemaStorage;
 import org.apache.pulsar.common.protocol.schema.StoredSchema;
 import org.apache.pulsar.common.schema.SchemaType;

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

@@ -21,6 +21,7 @@
 import java.nio.ByteBuffer;
 import org.apache.bookkeeper.client.api.BKException;
 import org.apache.pulsar.broker.PulsarService;
+import org.apache.pulsar.common.schema.LongSchemaVersion;
 import org.apache.pulsar.zookeeper.LocalZooKeeperCache;
 import org.testng.annotations.Test;
 

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

@@ -41,6 +41,7 @@
 import org.apache.pulsar.broker.service.schema.SchemaRegistry.SchemaAndMetadata;
 import org.apache.pulsar.common.policies.data.SchemaCompatibilityStrategy;
 import org.apache.pulsar.common.protocol.schema.SchemaData;
+import org.apache.pulsar.common.schema.LongSchemaVersion;
 import org.apache.pulsar.common.schema.SchemaType;
 import org.apache.pulsar.common.protocol.schema.SchemaVersion;
 import org.testng.annotations.AfterMethod;

pulsar-broker/src/test/java/org/apache/pulsar/client/api/SimpleSchemaTest.java

@@ -30,7 +30,7 @@
 
 import org.apache.avro.reflect.ReflectData;
 import org.apache.avro.Schema.Parser;
-import org.apache.pulsar.broker.service.schema.LongSchemaVersion;
+import org.apache.pulsar.common.schema.LongSchemaVersion;
 import org.apache.pulsar.client.admin.PulsarAdminException;
 import org.apache.pulsar.client.api.PulsarClientException.IncompatibleSchemaException;
 import org.apache.pulsar.client.api.PulsarClientException.InvalidMessageException;

pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/AbstractSchema.java

@@ -22,7 +22,7 @@
 import org.apache.pulsar.client.api.Schema;
 import org.apache.pulsar.client.api.SchemaSerializationException;
 
-abstract class AbstractSchema<T> implements Schema<T> {
+public abstract class AbstractSchema<T> implements Schema<T> {
 
     /**
      * Check if the message read able length length is a valid object for this schema.
@@ -47,7 +47,7 @@ void validate(ByteBuf byteBuf) {
      *            the byte buffer to decode
      * @return the deserialized object
      */
-    abstract T decode(ByteBuf byteBuf);
+    public abstract T decode(ByteBuf byteBuf);
     /**
      * Decode a byteBuf into an object using a given version.
      *
@@ -57,7 +57,7 @@ void validate(ByteBuf byteBuf) {
      *            the schema version to decode the object. null indicates using latest version.
      * @return the deserialized object
      */
-    T decode(ByteBuf byteBuf, byte[] schemaVersion) {
+    public T decode(ByteBuf byteBuf, byte[] schemaVersion) {
         // ignore version by default (most of the primitive schema implementations ignore schema version)
         return decode(byteBuf);
     }

pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/StructSchema.java

@@ -103,7 +103,8 @@ public T decode(byte[] bytes) {
     @Override
     public T decode(byte[] bytes, byte[] schemaVersion) {
         try {
-            return readerCache.get(BytesSchemaVersion.of(schemaVersion)).read(bytes);
+            return schemaVersion == null ? decode(bytes) :
+                    readerCache.get(BytesSchemaVersion.of(schemaVersion)).read(bytes);
         } catch (ExecutionException | AvroTypeException e) {
             if (e instanceof AvroTypeException) {
                 throw new SchemaSerializationException(e);
@@ -122,7 +123,8 @@ public T decode(ByteBuf byteBuf) {
     @Override
     public T decode(ByteBuf byteBuf, byte[] schemaVersion) {
         try {
-            return readerCache.get(BytesSchemaVersion.of(schemaVersion)).read(new ByteBufInputStream(byteBuf));
+            return schemaVersion == null ? decode(byteBuf) :
+                    readerCache.get(BytesSchemaVersion.of(schemaVersion)).read(new ByteBufInputStream(byteBuf));
         } catch (ExecutionException e) {
             LOG.error("Can't get generic schema for topic {} schema version {}",
                     schemaInfoProvider.getTopicName(), Hex.encodeHexString(schemaVersion), e);
@@ -180,7 +182,7 @@ protected static org.apache.avro.Schema parseAvroSchema(String schemaJson) {
         return parser.parse(schemaJson);
     }
 
-    protected static <T> SchemaInfo parseSchemaInfo(SchemaDefinition<T> schemaDefinition, SchemaType schemaType) {
+    public static <T> SchemaInfo parseSchemaInfo(SchemaDefinition<T> schemaDefinition, SchemaType schemaType) {
         return SchemaInfo.builder()
                 .schema(createAvroSchema(schemaDefinition).toString().getBytes(UTF_8))
                 .properties(schemaDefinition.getProperties())

pulsar-common/src/main/java/org/apache/pulsar/common/api/raw/RawMessage.java

@@ -102,6 +102,13 @@ public interface RawMessage {
      */
     Optional<String> getKey();
 
+    /**
+     * Get the schema verison of the message.
+     *
+     * @return the schema version of the message
+     */
+    byte[] getSchemaVersion();
+
     /**
      * Get byteBuf of the key.
      *
@@ -115,5 +122,4 @@ public interface RawMessage {
      * @return true if the key is base64 encoded, false otherwise
      */
     boolean hasBase64EncodedKey();
-
 }

pulsar-common/src/main/java/org/apache/pulsar/common/api/raw/RawMessageImpl.java

@@ -143,6 +143,14 @@ public Optional<String> getKey() {
     }
 
     @Override
+    public byte[] getSchemaVersion() {
+        if (msgMetadata != null && msgMetadata.get().hasSchemaVersion()) {
+            return msgMetadata.get().getSchemaVersion().toByteArray();
+        } else {
+            return null;
+        }
+    }
+
     public Optional<ByteBuf> getKeyBytes() {
         if (getKey().isPresent()) {
             if (hasBase64EncodedKey()) {