Add the multi version schema support #3876

congbobo184 · 2019-03-21T09:24:07Z

Motivation

Fix #3742
In order to decode the message correctly by AVRO schema, we need to know the schema what the message is.

Modification

For structSchema add reader and writer

reader is used to decode message
writer is used to encode message

Modifications

Every schema what implements the structSchema, add reader and writer they only have. Message encode by the write method in writer and decode by the read method in reader.
A cache in StructSchema for readers of every different version, so we only need to get the reader by schema version. The schema version is stored in message.

Verifying this change

Add multiVersionSchema

Does this pull request potentially affect one of the following parts:

If yes was chosen, please highlight the changes

Dependencies (does it add or upgrade a dependency): (no)

The public API: (no)
The schema: (yes)
The default values of configurations: (no)
The wire protocol: (no)
The rest endpoints: (no)
The admin cli options: (no)
Anything that affects deployment: (no)

sijie

@congbobo184

The implementation here is too customized to AvroSchema. I don't think it is a good implementation. Ideally I think multi-versioning should be handled by StructSchema class. For StructSchema, you need a writer schema provider and a reader schema provider. By default both writer and reader schema provider returns the schema generated by the pojo.

When a schema is used in producer, pulsar will fetch the latest schema from broker, and set reader schema to the broker side schema. so it has writer schema (generated from pojo) and reader schema (read from broker) to serialize the data.

When a schema is used in consumer, the writer schema provider will be the multi-version schema provider (which reads and caches schemas from broker), and the reader schema is the schema generated from POJO.

We should try to do this in an abstract way. So all the subclasses of StructSchema will have a writer schema and a reader schema for serializing and deserializing data.

sijie · 2019-03-21T09:28:30Z

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

@@ -83,6 +84,7 @@
    private final ConnectionPool cnxPool;
    private final Timer timer;
    private final ExecutorProvider externalExecutorProvider;
+    private Map<String, MultiVersionSchema> multiVersionSchemaCatch = new HashMap<>();


you mean Cache? multiVersionSchemaCache?

sijie · 2019-03-21T09:30:13Z

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

@@ -152,6 +155,17 @@
    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 instanceof AvroSchema && schema.supportSchemaVersioning()){
+            Map<String, MultiVersionSchema> multiVersionSchemaCatch = client.getMultiVersionSchemaCatch();
+            if(multiVersionSchemaCatch.get(topic) != null){


Suggested change

if(multiVersionSchemaCatch.get(topic) != null){

MultiVersionSchema schema = multiVersionSchemaCache.get(topic);

if(schema == null) {

...

}

sijie · 2019-03-21T09:38:20Z

tests/integration/src/test/java/org/apache/pulsar/tests/integration/schema/Schemas.java

+        private String name;
+        private int age;
+        @AvroDefault("\"man\"")
+        private String sex;


use gender (male, and female)

codelipenghui · 2019-03-26T06:21:52Z

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

@@ -152,6 +154,16 @@
    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 instanceof AvroSchema){


Why not schema.supportSchemaVersioning() ?

codelipenghui · 2019-03-26T06:36:50Z

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

+            if(schemaFromCache == null){
+                ((AvroSchema<T>) schema).setSchemaProvider(new MultiVersionGenericSchemaProvider(TopicName.get(topic), client));
+                supportSchemaVersioningSchemaCache.put(topic, schema);
+            }else{


And some spaces

congbobo184 · 2019-03-26T10:29:34Z

run Integration Tests

congbobo184 · 2019-03-27T01:26:40Z

run java8 tests

congbobo184 · 2019-03-27T07:06:12Z

run java8 tests

sijie · 2019-03-27T07:17:11Z

@congbobo184 can you rebase your PR to latest master?

congbobo184 · 2019-03-27T07:52:49Z

@congbobo184 can you rebase your PR to latest master?

OK,I'will do it

…chemaversion # Conflicts: # pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/AvroSchema.java

congbobo184 · 2019-03-27T10:07:35Z

run java8 tests

sijie

@congbobo184 overall looks much clearer now. However I think we still need to push the reader and writer schema logic to StructSchema.

sijie · 2019-03-27T15:51:58Z

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

@@ -31,6 +31,12 @@
     * @param schemaVersion schema version
     * @return schema instance of the provided <tt>schemaVersion</tt>
     */
-    Schema<T> getSchema(byte[] schemaVersion);
+    Schema<T> getVersionSchema(byte[] schemaVersion);


getSchema is probably okay. or we can call it getSchemaByVersion

sijie · 2019-03-27T15:52:13Z

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

+     *
+     * @return the current schema
+     */
+    Schema<T> getCurrentSchema() throws InterruptedException;


maybe getLatestSchema?

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

sijie · 2019-03-27T15:55:31Z

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

@@ -83,6 +83,7 @@
    private final ConnectionPool cnxPool;
    private final Timer timer;
    private final ExecutorProvider externalExecutorProvider;
+    private Map<String, Schema> supportSchemaVersioningSchemaCache = new HashMap<>();


either use a concurrent structure or you should synchronize on accessing supportSchemaVersioningSchemaCache in ConsumerImpl.

sijie · 2019-03-27T15:57:40Z

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

    private static final ThreadLocal<BinaryDecoder> decoders =
            new ThreadLocal<>();
+    private boolean supportSchemaVersioning;


I still think the code change here should be pushed to StructSchema. because reader and writer schemas should be applied to all struct schemas, although only Avro can use them now

All right, it should push to StructSchema

sijie · 2019-03-27T16:00:47Z

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

@@ -152,6 +154,16 @@
    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()) {


actually I am bit confused about the logic here. why do we need to cache this?

Event we want to cache this, we should cache the MultiVersionGenericSchemaProvider. but we shouldn't cache the schema. because user can pass in different type of schema instances.

congbobo184 · 2019-03-28T06:39:38Z

@congbobo184 overall looks much clearer now. However I think we still need to push the reader and writer schema logic to StructSchema.

I think we need to push the reader and writer, not reader schema and writer schema. Because, from the point of view of shcema, it shouldn't have the concept of the reader schema or writer schema.

sijie · 2019-03-28T06:43:45Z

I think we need to push the reader and writer, not reader schema and writer schema.

@congbobo184 yes

…chemaversion

congbobo184 · 2019-04-01T10:21:42Z

run java8 tests

…chemaversion

congbobo184 · 2019-04-22T16:14:05Z

run cpp tests

sijie

@congbobo184 Really great contribution!

I think it is almost ready to go. However I think there is one place you didn't handle the exception very well, can you please take a look and fix it?

sijie · 2019-04-23T17:41:38Z

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

+        try {
+            client.preProcessSchemaBeforeSubscribe(client, schema, topicName);
+        } catch (Throwable t) {
+            throw new RuntimeException(t);


@congbobo184 you should fail subscribeResult here, no?

congbobo184 · 2019-04-24T05:33:10Z

run java8 tests

sijie · 2019-04-24T06:02:13Z

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

+        try {
+            client.preProcessSchemaBeforeSubscribe(client, schema, topicName);
+        } catch (Throwable t) {
+            subscribeResult.completeExceptionally(t);


@congbobo184 don't you need to return here?

yes, It should be

congbobo184 · 2019-04-24T09:50:14Z

run java8 Tests

codelipenghui · 2019-04-26T03:49:38Z

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

+    /**
+     * serialize the object into bytes
+     *
+     * @param pojo the pojo


Is the parameter name here correct?

I will change it named message.

jiazhai · 2019-04-26T04:15:13Z

pulsar-client/src/test/java/org/apache/pulsar/client/impl/schema/SchemaBuilderTest.java

-import org.apache.pulsar.client.api.schema.GenericSchema;
-import org.apache.pulsar.client.api.schema.RecordSchemaBuilder;
-import org.apache.pulsar.client.api.schema.SchemaBuilder;
+import org.apache.pulsar.client.api.schema.*;


nit: We should avoid this "*" kind import

jiazhai · 2019-04-26T04:15:44Z

...ient/src/test/java/org/apache/pulsar/client/impl/schema/SupportVersioningAvroSchemaTest.java

+import org.testng.annotations.Test;
+
+import static org.mockito.Matchers.any;
+import static org.mockito.Mockito.*;


nit: same here, avoid * kind import

jiazhai · 2019-04-26T04:26:39Z

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

@@ -230,7 +230,7 @@ public boolean isExpired(int messageTTLInSeconds) {

    @Override
    public byte[] getSchemaVersion() {
-        if (msgMetadataBuilder.hasSchemaVersion()) {
+        if (msgMetadataBuilder != null && msgMetadataBuilder.hasSchemaVersion() ) {


nit: remove the addition space " " between ") )"

jiazhai

+1, over all lgtm. @congbobo184 Added some minor comments

* modify the test import * * remove space

congbobo184 · 2019-04-26T06:47:47Z

+1, over all lgtm. @congbobo184 Added some minor comments

OK,I will change these

sijie · 2019-04-28T04:06:08Z

well done @congbobo184 ! this change makes pulsar be able to handle schema versions correctly!

congbobo184 · 2019-04-28T06:20:11Z

well done @congbobo184 ! this change makes pulsar be able to handle schema versions correctly!

@sijie Thank you for giving me a lot of programming advice.

Motivation Pulsar 2.4.0 Added schema versioning to support multi version messages produce and consume #3876 #3670 #4211 #4325 #4548. but the doc is not updated accordingly. Modifications Update the schema version in the pulsar registry doc for releases 2.4.0/2.4.1/2.4.2.

Motivation Pulsar 2.4.0 Added schema versioning to support multi version messages produce and consume apache#3876 apache#3670 apache#4211 apache#4325 apache#4548. but the doc is not updated accordingly. Modifications Update the schema version in the pulsar registry doc for releases 2.4.0/2.4.1/2.4.2.

add the multi version schema support

54d58c0

codelipenghui requested review from jerrypeng, sijie, merlimat, jiazhai and codelipenghui and removed request for sijie March 21, 2019 09:33

congbobo added 2 commits March 21, 2019 17:38

recover the code format

2776e32

recover the code format

9714c9d

sijie requested changes Mar 21, 2019

View reviewed changes

congbobo added 2 commits March 21, 2019 18:18

Modify the cache name

9447588

set schema provider in struct schema

b01f509

codelipenghui reviewed Mar 26, 2019

View reviewed changes

congbobo added 2 commits March 26, 2019 16:54

Modify the jude type

5dae355

Modify the judge condition

ee531dc

Merge remote-tracking branch 'apache/master' into schema_analysis_bys…

cea87d0

…chemaversion # Conflicts: # pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/AvroSchema.java

sijie requested changes Mar 27, 2019

View reviewed changes

merlimat added this to the 2.4.0 milestone Mar 30, 2019

congbobo added 2 commits April 1, 2019 11:10

add the schema reader and writer

70ca1fe

Merge remote-tracking branch 'apache/master' into schema_analysis_bys…

c491e42

…chemaversion

congbobo added 4 commits April 22, 2019 11:11

* modify the location for schema for preProcessSchemaBeforeSubscribe

2f87051

* modify the location of add schemaInfoProvider

e210c86

* modify the topic name to get schema Info in MultiTopicsConsumerImpl

564dcee

Merge remote-tracking branch 'apache/master' into schema_analysis_bys…

66819a7

…chemaversion

sijie approved these changes Apr 23, 2019

View reviewed changes

sijie assigned congbobo184 Apr 23, 2019

sijie added component/schemaregistry type/enhancement The enhancements for the existing features or docs. e.g. reduce memory usage of the delayed messages labels Apr 23, 2019

modify the exception in multi topic for getting provider

cd5b485

sijie reviewed Apr 24, 2019

View reviewed changes

* add the return

0691bf3

sijie approved these changes Apr 25, 2019

View reviewed changes

codelipenghui reviewed Apr 26, 2019

View reviewed changes

jiazhai reviewed Apr 26, 2019

View reviewed changes

jiazhai approved these changes Apr 26, 2019

View reviewed changes

* modify the parameter name for schema writer method

091f69c

* modify the test import * * remove space

codelipenghui approved these changes Apr 28, 2019

View reviewed changes

sijie merged commit d5ff082 into apache:master Apr 28, 2019

Huanli-Meng mentioned this pull request Jul 1, 2020

[Doc]--Added schema versioning to support multi version messages produce and consume #7417

Merged

codelipenghui mentioned this pull request Nov 6, 2020

Producer Consumer design #8417

Closed

-            if(multiVersionSchemaCatch.get(topic) != null){
+            MultiVersionSchema schema = multiVersionSchemaCache.get(topic);
+            if(schema == null) {
+               ...
+            }

Add the multi version schema support #3876

Add the multi version schema support #3876

Conversation

congbobo184 commented Mar 21, 2019 • edited by jiazhai

Motivation

Modification

Modifications

Verifying this change

Does this pull request potentially affect one of the following parts:

Dependencies (does it add or upgrade a dependency): (no)

sijie left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

congbobo184 commented Mar 26, 2019

congbobo184 commented Mar 27, 2019

congbobo184 commented Mar 27, 2019

sijie commented Mar 27, 2019

congbobo184 commented Mar 27, 2019

congbobo184 commented Mar 27, 2019

sijie left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

congbobo184 commented Mar 28, 2019

sijie commented Mar 28, 2019

congbobo184 commented Apr 1, 2019

congbobo184 commented Apr 22, 2019

sijie left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

congbobo184 commented Apr 24, 2019

Choose a reason for hiding this comment

Choose a reason for hiding this comment

congbobo184 commented Apr 24, 2019

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

jiazhai left a comment

Choose a reason for hiding this comment

congbobo184 commented Apr 26, 2019

sijie commented Apr 28, 2019

congbobo184 commented Apr 28, 2019 • edited

congbobo184 commented Mar 21, 2019 •

edited by jiazhai

congbobo184 commented Apr 28, 2019 •

edited