[Schema] Support consume multiple schema types messages by AutoConsumeSchema #10604

gaoran10 · 2021-05-17T03:33:09Z

Based on the PR #10573

Motivation

Support consuming multiple schema types messages by AutoConsumeSchema.

Modifications

Describe the modifications you've done.

Verifying this change

Add a new unit test to verify consuming multiple schema type messages by AutoConsumeSchema.

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: (no)
The default values of configurations: (no)
The wire protocol: (no)
The rest endpoints: (no)
The admin cli options: (no)
Anything that affects deployment: (no)

eolivelli

nice work
it is a good complement for all of the schema related enhancements we delivered in 2.8.0

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

eolivelli

I have merged #10573
please rebase.

eolivelli

LGTM

codelipenghui · 2021-05-18T00:01:15Z

Error:  Tests run: 6, Failures: 1, Errors: 0, Skipped: 5, Time elapsed: 0.273 s <<< FAILURE! - in org.apache.pulsar.client.impl.schema.generic.GenericSchemaTest
Error:  testAutoAvroSchema(org.apache.pulsar.client.impl.schema.generic.GenericSchemaTest)  Time elapsed: 0.034 s  <<< FAILURE!
java.lang.NullPointerException
	at org.apache.pulsar.client.impl.schema.generic.GenericSchemaTest.verifyFooRecord(GenericSchemaTest.java:191)
	at org.apache.pulsar.client.impl.schema.generic.GenericSchemaTest.testEncodeAndDecodeGenericRecord(GenericSchemaTest.java:118)
	at org.apache.pulsar.client.impl.schema.generic.GenericSchemaTest.testAutoAvroSchema(GenericSchemaTest.java:81)
	at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
	at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
	at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
	at java.base/java.lang.reflect.Method.invoke(Method.java:566)
	at org.testng.internal.MethodInvocationHelper.invokeMethod(MethodInvocationHelper.java:132)
	at org.testng.internal.InvokeMethodRunnable.runOne(InvokeMethodRunnable.java:45)
	at org.testng.internal.InvokeMethodRunnable.call(InvokeMethodRunnable.java:73)
	at org.testng.internal.InvokeMethodRunnable.call(InvokeMethodRunnable.java:11)
	at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)
	at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
	at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)
	at java.base/java.lang.Thread.run(Thread.java:829)

eolivelli · 2021-05-18T06:30:24Z

...st/java/org/apache/pulsar/client/impl/schema/generic/MultiVersionSchemaInfoProviderTest.java

@@ -61,7 +62,7 @@ public void testGetSchema() throws Exception {
                        any(TopicName.class),
                        any(byte[].class)))
                .thenReturn(completableFuture);
-        SchemaInfo schemaInfoByVersion = schemaProvider.getSchemaByVersion(new byte[0]).get();
+        SchemaInfo schemaInfoByVersion = schemaProvider.getSchemaByVersion(new LongSchemaVersion(0).bytes()).get();


it looks like we are changing the behaviour.
can this be a problem ?
do we need to add some compatibility layer ?

The change of the class MultiVersionSchemaInfoProvider is not necessary and I revert this test. Thanks.

codelipenghui · 2021-05-19T01:04:57Z

/pulsarbot run-failure-checks

eolivelli

LGTM

I left one suggestion for the test case, PTAL

eolivelli · 2021-05-19T12:10:31Z

pulsar-broker/src/test/java/org/apache/pulsar/schema/SchemaTest.java

+                Assert.assertEquals(genericRecord.get("id"), 10);
+                break;
+            case "k_one_v_three_schema_separate":
+                kv = (KeyValue<?, ?>) nativeObject;


this should be a KeyValue<GenericRecord, GenericRecord>

Ok, I'll fix.

eolivelli · 2021-05-19T12:10:40Z

pulsar-broker/src/test/java/org/apache/pulsar/schema/SchemaTest.java

+                kv = (KeyValue<?, ?>) nativeObject;
+                jsonNode = ((GenericJsonRecord) kv.getKey()).getJsonNode();
+                Assert.assertEquals(jsonNode.get("id").intValue(), 1);
+                jsonNode = ((GenericJsonRecord) kv.getValue()).getJsonNode();


what about kv.getValue().getNativeObject() ?

It seems that using the getNativeObject still needs to constraint convert to JsonNode.

…ema AutoConsumeSchema

gaoran10 · 2021-05-21T13:35:34Z

There are still two problems that need to be resolved when using AutoConsumeSchema to consume key-value schema messages.

The keyValue schema data encoding does not consider the KeyValueEncodingType, so if the key and value schema are some but not some KeyValueEncodingType, this will cause an error when decoding message data.

For example, these two schemas belong to one topic, there will be only one schema version due to they have the same schema data.

Schema.KeyValue(Schema.JSON(Schemas.PersonOne.class),
                        Schema.JSON(Schemas.PersonFour.class), KeyValueEncodingType.SEPARATED)
Schema.KeyValue(Schema.JSON(Schemas.PersonOne.class),
                        Schema.JSON(Schemas.PersonFour.class), KeyValueEncodingType.INLINE)

Primitive schemas have empty schema data, this will cause the KeyValue schema data doesn't distinguish different primitive schemas.

For example.

Schema.KeyValue(Schema.INT32, Schema.JSON(Schemas.PersonThree.class), KeyValueEncodingType.SEPARATED)
Schema.KeyValue(Schema.BOOL, Schema.JSON(Schemas.PersonThree.class), KeyValueEncodingType.SEPARATED)

Maybe we could adjust the key-value schema encoding logic.

Fix these two problems in the next PR.

eolivelli · 2021-05-21T14:41:28Z

@gaoran10 I would postpone fixing those problems to a follow up patch
this patch is already quite big.

eolivelli · 2021-05-21T14:43:34Z

pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/LookupProxyHandler.java

@@ -357,6 +358,11 @@ public void handleGetSchema(CommandGetSchema commandGetSchema) {
        final long clientRequestId = commandGetSchema.getRequestId();
        String serviceUrl = getBrokerServiceUrl(clientRequestId);
        String topic = commandGetSchema.getTopic();
+        Optional<SchemaVersion> tempVersion = Optional.empty();


what about:

final Optional<SchemaVersion> schemaVersion; if (commandGetSchema.hasSchemaVersion()) { schemaVersion = Optional.of(commandGetSchema.getSchemaVersion()).map(BytesSchemaVersion::of); } else { schemaVersion = Optional.empty() }

Sounds good!

congbobo184 · 2021-05-23T04:37:01Z

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

+        SchemaVersion sv = BytesSchemaVersion.of(schemaVersion);
+        fetchSchemaIfNeeded(sv);
+        ensureSchemaInitialized(sv);
+        return adapt(schemaMap.get(sv).decode(bytes, schemaVersion), schemaVersion);


may don't need decode(bytes, schemaVersion), only use decode(bytes) best?

Yes, good idea!

congbobo184 · 2021-05-23T05:02:53Z

pulsar-broker/src/test/java/org/apache/pulsar/schema/SchemaTest.java

+                Assert.assertEquals(jsonNode.get("age").intValue(), 18);
+                break;
+            case "avro_one_schema":
+                org.apache.avro.generic.GenericRecord genericRecord =


does this type is PersonOne not GenericAvroRecord is right?

It seems that if using the Schema.AUTO_CONSUME(), the reader doesn't know the Pojo PersonOne so the decoding result is org.apache.avro.generic.GenericRecord.

…eSchema (apache#10604) Based on the PR apache#10573 Support consuming multiple schema types messages by AutoConsumeSchema. (cherry picked from commit f5e10a9)

…toConsumeSchema (apache#10604)" This reverts commit 3398e3f.

…eSchema (apache#10604) Based on the PR apache#10573 ### Motivation Support consuming multiple schema types messages by AutoConsumeSchema.

eolivelli requested changes May 17, 2021

View reviewed changes

eolivelli approved these changes May 17, 2021

View reviewed changes

codelipenghui assigned gaoran10 May 17, 2021

codelipenghui added this to the 2.8.0 milestone May 17, 2021

codelipenghui added the component/schemaregistry label May 17, 2021

codelipenghui approved these changes May 17, 2021

View reviewed changes

eolivelli approved these changes May 17, 2021

View reviewed changes

eolivelli reviewed May 18, 2021

View reviewed changes

eolivelli approved these changes May 19, 2021

View reviewed changes

gaoran10 added 13 commits May 20, 2021 09:36

support consume multiple schema type message data by auto consume sch…

82ee345

…ema AutoConsumeSchema

fix comments

e003d51

fix test

4849d42

fix test

2b5a97c

fix test

3d0362f

fix test

cf33346

fix

7722496

fix produce key value schema data by auto produce schema

42f3aaa

fix comment

33a949f

support produce multiple schema by AutoProduceBytesSchema

15a1f20

fix test

45f5d0a

fix

bd13e30

fix

3a801e0

gaoran10 force-pushed the consume-multiple-schema-messages branch from 864a188 to 3a801e0 Compare May 20, 2021 02:20

fix test

c10af7c

eolivelli reviewed May 21, 2021

View reviewed changes

fix comments

1483388

codelipenghui requested review from sijie, merlimat and congbobo184 May 23, 2021 02:30

congbobo184 reviewed May 23, 2021

View reviewed changes

fix comments

7cf5440

codelipenghui merged commit f5e10a9 into apache:master May 23, 2021

eolivelli added a commit to datastax/pulsar that referenced this pull request Jun 7, 2021

Revert "[Schema] Support consume multiple schema types messages by Au…

1d34fa6

…toConsumeSchema (apache#10604)" This reverts commit 3398e3f.

gaoran10 deleted the consume-multiple-schema-messages branch April 8, 2022 08:14

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[Schema] Support consume multiple schema types messages by AutoConsumeSchema #10604

[Schema] Support consume multiple schema types messages by AutoConsumeSchema #10604

gaoran10 commented May 17, 2021

eolivelli left a comment

eolivelli left a comment

eolivelli left a comment

codelipenghui commented May 18, 2021

eolivelli May 18, 2021

gaoran10 May 18, 2021

codelipenghui commented May 19, 2021

eolivelli left a comment

eolivelli May 19, 2021

gaoran10 May 20, 2021

eolivelli May 19, 2021

gaoran10 May 20, 2021

gaoran10 commented May 21, 2021 •

edited

Loading

eolivelli commented May 21, 2021

eolivelli May 21, 2021

gaoran10 May 21, 2021

congbobo184 May 23, 2021 •

edited

Loading

gaoran10 May 23, 2021

congbobo184 May 23, 2021

gaoran10 May 23, 2021

[Schema] Support consume multiple schema types messages by AutoConsumeSchema #10604

[Schema] Support consume multiple schema types messages by AutoConsumeSchema #10604

Conversation

gaoran10 commented May 17, 2021

Motivation

Modifications

Verifying this change

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

eolivelli left a comment

Choose a reason for hiding this comment

eolivelli left a comment

Choose a reason for hiding this comment

eolivelli left a comment

Choose a reason for hiding this comment

codelipenghui commented May 18, 2021

Choose a reason for hiding this comment

Choose a reason for hiding this comment

codelipenghui commented May 19, 2021

eolivelli 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

gaoran10 commented May 21, 2021 • edited Loading

eolivelli commented May 21, 2021

Choose a reason for hiding this comment

Choose a reason for hiding this comment

congbobo184 May 23, 2021 • edited Loading

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

gaoran10 commented May 21, 2021 •

edited

Loading

congbobo184 May 23, 2021 •

edited

Loading