Add the multi version schema support

pulsar-client-api/src/main/java/org/apache/pulsar/client/api/schema/SchemaDefinition.java

@@ -61,4 +61,11 @@ static  <T> SchemaDefinitionBuilder<T> builder() {
      * @return pojo schema
      */
     public Class<T> getPojo();
+
+    /**
+     * Get supportSchemaVersioning schema definition
+     *
+     * @return the flag of supportSchemaVersioning
+     */
+    public boolean getSupportSchemaVersioning();
 }

pulsar-client-api/src/main/java/org/apache/pulsar/client/api/schema/SchemaDefinitionBuilder.java

@@ -49,7 +49,7 @@ public interface SchemaDefinitionBuilder<T> {
      * @param key property key
      * @param value property value
      *
-     * @return record schema definition
+     * @return schema definition builder
      */
     SchemaDefinitionBuilder<T> addProperty(String key, String value);
 
@@ -58,7 +58,7 @@ public interface SchemaDefinitionBuilder<T> {
      *
      * @param pojo pojo schema definition
      *
-     * @return record schema definition
+     * @return schema definition builder
      */
     SchemaDefinitionBuilder<T> withPojo(Class pojo);
 
@@ -67,10 +67,19 @@ public interface SchemaDefinitionBuilder<T> {
      *
      * @param jsonDefinition json schema definition
      *
-     * @return record schema definition
+     * @return schema definition builder
      */
     SchemaDefinitionBuilder<T> withJsonDef(String jsonDefinition);
 
+    /**
+     * Set schema whether decode by schema version
+     *
+     * @param supportSchemaVersioning decode by version
+     *
+     * @return schema definition builder
+     */
+    SchemaDefinitionBuilder<T> withSupportSchemaVersioning(boolean supportSchemaVersioning);
+
     /**
      * Build the schema definition.
      *

pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/generic/SchemaProvider.java → pulsar-client-api/src/main/java/org/apache/pulsar/client/api/schema/SchemaProvider.java

@@ -16,7 +16,7 @@
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.pulsar.client.impl.schema.generic;
+package org.apache.pulsar.client.api.schema;
 
 import org.apache.pulsar.client.api.Schema;
 
@@ -31,6 +31,12 @@ public interface SchemaProvider<T> {
      * @param schemaVersion schema version
      * @return schema instance of the provided <tt>schemaVersion</tt>
      */
-    Schema<T> getSchema(byte[] schemaVersion);
+    Schema<T> getVersionSchema(byte[] schemaVersion);
+    /**
+     * Retrieve the current schema.
+     *
+     * @return the current schema
+     */
+    Schema<T> getCurrentSchema() throws InterruptedException;
 
 }

pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java

@@ -63,6 +63,8 @@
 import org.apache.pulsar.client.api.SubscriptionInitialPosition;
 import org.apache.pulsar.client.api.SubscriptionType;
 import org.apache.pulsar.client.impl.conf.ConsumerConfigurationData;
+import org.apache.pulsar.client.impl.schema.AvroSchema;
+import org.apache.pulsar.client.impl.schema.generic.MultiVersionGenericSchemaProvider;
 import org.apache.pulsar.common.api.Commands;
 import org.apache.pulsar.common.api.EncryptionContext;
 import org.apache.pulsar.common.api.EncryptionContext.EncryptionKey;
@@ -152,6 +154,16 @@ enum SubscriptionMode {
     static <T> ConsumerImpl<T> newConsumerImpl(PulsarClientImpl client, String topic, ConsumerConfigurationData<T> conf,
                  ExecutorService listenerExecutor, int partitionIndex, CompletableFuture<Consumer<T>> subscribeFuture,
                  SubscriptionMode subscriptionMode, MessageId startMessageId, Schema<T> schema, ConsumerInterceptors<T> interceptors) {
+        if (schema != null && schema.supportSchemaVersioning()) {
+            Map<String, Schema> supportSchemaVersioningSchemaCache = client.getSupportSchemaVersioningSchemaCache();
+            Schema<T> schemaFromCache = supportSchemaVersioningSchemaCache.get(topic);
+            if (schemaFromCache == null) {
+                ((AvroSchema<T>) schema).setSchemaProvider(new MultiVersionGenericSchemaProvider(TopicName.get(topic), client));
+                supportSchemaVersioningSchemaCache.put(topic, schema);
+            } else {
+                schema = schemaFromCache;
+            }
+        }
         if (conf.getReceiverQueueSize() == 0) {
             return new ZeroQueueConsumerImpl<>(client, topic, conf, listenerExecutor, partitionIndex, subscribeFuture,
                     subscriptionMode, startMessageId, schema, interceptors);

pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientImpl.java

@@ -83,6 +83,7 @@ public class PulsarClientImpl implements PulsarClient {
     private final ConnectionPool cnxPool;
     private final Timer timer;
     private final ExecutorProvider externalExecutorProvider;
+    private Map<String, Schema> supportSchemaVersioningSchemaCache = new HashMap<>();
 
     enum State {
         Open, Closing, Closed
@@ -736,4 +737,8 @@ private static Mode convertRegexSubscriptionMode(RegexSubscriptionMode regexSubs
             return null;
         }
     }
+
+    public Map<String, Schema> getSupportSchemaVersioningSchemaCache(){
+        return this.supportSchemaVersioningSchemaCache;
+    }
 }

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

@@ -18,6 +18,9 @@
  */
 package org.apache.pulsar.client.impl.schema;
 
+import com.google.common.cache.CacheBuilder;
+import com.google.common.cache.CacheLoader;
+import com.google.common.cache.LoadingCache;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.avro.Conversions;
 import org.apache.avro.data.TimeConversions;
@@ -30,26 +33,42 @@
 import org.apache.avro.reflect.ReflectDatumWriter;
 import org.apache.pulsar.client.api.SchemaSerializationException;
 import org.apache.pulsar.client.api.schema.SchemaDefinition;
+import org.apache.pulsar.client.impl.schema.generic.GenericAvroSchema;
+import org.apache.pulsar.client.impl.schema.generic.MultiVersionGenericSchemaProvider;
 import org.apache.pulsar.common.schema.SchemaInfo;
 import org.apache.pulsar.common.schema.SchemaType;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import java.io.ByteArrayOutputStream;
 import java.io.IOException;
+import java.nio.charset.StandardCharsets;
 import java.util.Map;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
 
 /**
  * An AVRO schema implementation.
  */
 @Slf4j
 public class AvroSchema<T> extends StructSchema<T> {
+    private static final Logger LOG = LoggerFactory.getLogger(AvroSchema.class);
 
     private ReflectDatumWriter<T> datumWriter;
-    private ReflectDatumReader<T> reader;
+    private ReflectDatumReader<T> datumReader;
     private BinaryEncoder encoder;
     private ByteArrayOutputStream byteArrayOutputStream;
-
     private static final ThreadLocal<BinaryDecoder> decoders =
             new ThreadLocal<>();
+    private boolean supportSchemaVersioning;
+    private final LoadingCache<byte[], ReflectDatumReader<T>> cache = CacheBuilder.newBuilder().maximumSize(100000)
+            .expireAfterAccess(30, TimeUnit.MINUTES).build(new CacheLoader<byte[], ReflectDatumReader<T>>() {
+                @Override
+                public ReflectDatumReader<T> load(byte[] schemaVersion) throws Exception {
+                    return loadReader(schemaVersion);
+                }
+            });
+
 //      the aim to fix avro's bug
 //      https://issues.apache.org/jira/browse/AVRO-1891  bug address explain
 //      fix the avro logical type read and write
@@ -85,8 +104,9 @@ private AvroSchema(org.apache.avro.Schema schema,
             schemaDefinition.getProperties());
         this.byteArrayOutputStream = new ByteArrayOutputStream();
         this.encoder = EncoderFactory.get().binaryEncoder(this.byteArrayOutputStream, this.encoder);
-        this.datumWriter = new ReflectDatumWriter<>(this.schema);
-        this.reader = new ReflectDatumReader<>(this.schema);
+        this.datumWriter = new ReflectDatumWriter<>(schema);
+        this.datumReader = new ReflectDatumReader<>(schema);
+        this.supportSchemaVersioning = schemaDefinition.getSupportSchemaVersioning();
     }
 
     @Override
@@ -110,9 +130,27 @@ public T decode(byte[] bytes) {
             if (decoderFromCache == null) {
                 decoders.set(decoder);
             }
-            return reader.read(null, DecoderFactory.get().binaryDecoder(bytes, decoder));
+            return datumReader.read(null, DecoderFactory.get().binaryDecoder(bytes, decoder));
+        } catch (IOException e) {
+            throw new SchemaSerializationException(e);
+        }
+    }
+
+    @Override
+    public T decode(byte[] bytes, byte[] schemaVersion) {
+        try {
+            BinaryDecoder decoderFromCache = decoders.get();
+            BinaryDecoder decoder = DecoderFactory.get().binaryDecoder(bytes, decoderFromCache);
+            if (decoderFromCache == null) {
+                decoders.set(decoder);
+            }
+            return cache.get(schemaVersion).read(null, DecoderFactory.get().binaryDecoder(bytes, decoder));
         } catch (IOException e) {
             throw new SchemaSerializationException(e);
+        } catch (ExecutionException e) {
+            LOG.error("Can't get generic schema for topic {} schema version {}",
+                    ((MultiVersionGenericSchemaProvider)schemaProvider).getTopic().toString(), new String(schemaVersion, StandardCharsets.UTF_8), e);
+            return null;
         }
     }
 
@@ -135,4 +173,12 @@ public static <T> AvroSchema<T> of(Class<T> pojo, Map<String, String> properties
         return new AvroSchema<>(createAvroSchema(schemaDefinition), schemaDefinition);
     }
 
+    public boolean supportSchemaVersioning(){
+        return supportSchemaVersioning;
+    }
+
+    private ReflectDatumReader loadReader(byte[] schemaVersion)  {
+        return new ReflectDatumReader<T>(((GenericAvroSchema)schemaProvider.getVersionSchema(schemaVersion)).getAvroSchema(),schema);
+    }
+
 }

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

@@ -55,6 +55,11 @@ public class SchemaDefinitionBuilderImpl<T> implements SchemaDefinitionBuilder<T
      */
     private String jsonDef;
 
+    /**
+     * The flag of message decode whether by schema version
+     */
+    private boolean supportSchemaVersioning = false;
+
     @Override
     public SchemaDefinitionBuilder<T> withAlwaysAllowNull(boolean alwaysAllowNull) {
         this.alwaysAllowNull = alwaysAllowNull;
@@ -79,6 +84,11 @@ public SchemaDefinitionBuilder<T> withJsonDef(String jsonDef) {
         return this;
     }
 
+    @Override
+    public SchemaDefinitionBuilder<T> withSupportSchemaVersioning(boolean supportSchemaVersioning) {
+        this.supportSchemaVersioning = supportSchemaVersioning;
+        return this;
+    }
 
     @Override
     public SchemaDefinitionBuilder<T> withProperties(Map<String,String> properties) {
@@ -89,7 +99,7 @@ public SchemaDefinitionBuilder<T> withProperties(Map<String,String> properties)
     @Override
     public  SchemaDefinition<T> build() {
         properties.put(ALWAYS_ALLOW_NULL, this.alwaysAllowNull ? "true" : "false");
-        return new SchemaDefinitionImpl(clazz, jsonDef, alwaysAllowNull, properties);
+        return new SchemaDefinitionImpl(clazz, jsonDef, alwaysAllowNull, properties, supportSchemaVersioning);
 
     }
 }

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

@@ -21,7 +21,6 @@
 
 import org.apache.pulsar.client.api.schema.SchemaDefinition;
 
-import java.util.HashMap;
 import java.util.Map;
 
 /**
@@ -48,19 +47,21 @@ public class SchemaDefinitionImpl<T> implements SchemaDefinition<T>{
 
     private String jsonDef;
 
-    public SchemaDefinitionImpl(Class<T> pojo, String jsonDef, boolean alwaysAllowNull, Map<String,String> properties) {
+    private boolean supportSchemaVersioning;
+
+    public SchemaDefinitionImpl(Class<T> pojo, String jsonDef, boolean alwaysAllowNull, Map<String,String> properties, boolean supportSchemaVersioning) {
         this.alwaysAllowNull = alwaysAllowNull;
         this.properties = properties;
         this.jsonDef = jsonDef;
         this.pojo = pojo;
+        this.supportSchemaVersioning = supportSchemaVersioning;
     }
     /**
      * get schema whether always allow null or not
      *
      * @return schema always null or not
      */
     public boolean getAlwaysAllowNull() {
-
         return alwaysAllowNull;
     }
 
@@ -70,7 +71,6 @@ public boolean getAlwaysAllowNull() {
      * @return schema class
      */
     public String getJsonDef() {
-
         return jsonDef;
     }
     /**
@@ -83,16 +83,18 @@ public Class<T> getPojo() {
         return pojo;
     }
 
+    @Override
+    public boolean getSupportSchemaVersioning() {
+        return supportSchemaVersioning;
+    }
+
     /**
      * Get schema class
      *
      * @return schema class
      */
     public Map<String, String> getProperties() {
-
         return properties;
     }
 
-
-
 }

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

@@ -25,6 +25,7 @@
 import org.apache.avro.reflect.ReflectData;
 import org.apache.pulsar.client.api.Schema;
 import org.apache.pulsar.client.api.schema.SchemaDefinition;
+import org.apache.pulsar.client.api.schema.SchemaProvider;
 import org.apache.pulsar.common.schema.SchemaInfo;
 import org.apache.pulsar.common.schema.SchemaType;
 
@@ -42,6 +43,7 @@ abstract class StructSchema<T> implements Schema<T> {
 
     protected final org.apache.avro.Schema schema;
     protected final SchemaInfo schemaInfo;
+    protected SchemaProvider schemaProvider;
 
     protected StructSchema(SchemaType schemaType,
                            org.apache.avro.Schema schema,
@@ -54,7 +56,7 @@ protected StructSchema(SchemaType schemaType,
         this.schemaInfo.setProperties(properties);
     }
 
-    protected org.apache.avro.Schema getAvroSchema() {
+    public org.apache.avro.Schema getAvroSchema() {
         return schema;
     }
 
@@ -73,4 +75,8 @@ protected static org.apache.avro.Schema parseAvroSchema(String jsonDef) {
         return parser.parse(jsonDef);
     }
 
+    public void setSchemaProvider(SchemaProvider schemaProvider){
+        this.schemaProvider = schemaProvider;
+    }
+
 }

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

@@ -36,7 +36,7 @@
 /**
  * A generic avro schema.
  */
-class GenericAvroSchema extends GenericSchemaImpl {
+public class GenericAvroSchema extends GenericSchemaImpl {
 
     private final GenericDatumWriter<org.apache.avro.generic.GenericRecord> datumWriter;
     private BinaryEncoder encoder;

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

@@ -20,6 +20,7 @@
 
 import org.apache.pulsar.client.api.Schema;
 import org.apache.pulsar.client.api.schema.GenericRecord;
+import org.apache.pulsar.client.api.schema.SchemaProvider;
 import org.apache.pulsar.client.impl.schema.BytesSchema;
 import org.apache.pulsar.common.schema.SchemaInfo;
 
@@ -46,12 +47,12 @@ public boolean supportSchemaVersioning() {
 
     @Override
     public GenericRecord decode(byte[] bytes) {
-        return provider.getSchema(null).decode(bytes);
+        return provider.getVersionSchema(null).decode(bytes);
     }
 
     @Override
     public GenericRecord decode(byte[] bytes, byte[] schemaVersion) {
-        return provider.getSchema(schemaVersion).decode(bytes, schemaVersion);
+        return provider.getVersionSchema(schemaVersion).decode(bytes, schemaVersion);
     }
 
     @Override