[PIP-88] Replicate schemas across clusters #11441

codelipenghui · 2021-07-23T12:08:20Z

Motivation

Here is the proposal: https://github.com/apache/pulsar/wiki/PIP-88%3A-Replicate-schemas-across-multiple
For the implementation, we just need to set the correct SchemaInfo for the replicated message and using
the AutoProduceByte schema for the producer of the replicator.

Verifying this change

Added new tests to verify the consumer can consume messages with schema from the remote cluster.

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)

Here is the proposal: https://github.com/apache/pulsar/wiki/PIP-88%3A-Replicate-schemas-across-multiple For the implementation, we just need to set the correct SchemaInfo for the replicated message and using the AutoProduceByte schema for the producer of the replicator.

315157973 · 2021-07-26T03:17:10Z

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

@@ -64,6 +64,7 @@
    private ByteBuf payload;

    private Schema<T> schema;
+    private SchemaInfo schemaInfo;


I am a little confused about the modification.
We can get schemaInfo from schema now.
Can we use public static Schema<?> getSchema(SchemaInfo schemaInfo) in AutoConsumeSchema to convert schemaInfo to Schema

For the schema replication, we don't need to convert schemaInfo to schema for the local topic and then convert the schema to the schemaInfo for the remote topic. Or we only need to pass the schemaInfo to the message which need to replicate to the remote cluster.

We already have a schema, but we added a schemaInfo, which will make other developers very confused. If we can't reuse existing attributes, we can at least do this:

Make the new properties easier to understand, such as: naming modification, or adding comments, so that other developers can know schemaInfo only for replicator

Independent get/set

I second @315157973 concern.
We already have "getReaderSchema", why cannot we use that ?

eolivelli

the change looks good, apart the fact that we are adding that SchemaInfo field from MessageImpl

eolivelli · 2021-07-30T05:50:04Z

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

@@ -64,6 +64,7 @@
    private ByteBuf payload;

    private Schema<T> schema;
+    private SchemaInfo schemaInfo;


I second @315157973 concern.
We already have "getReaderSchema", why cannot we use that ?

eolivelli · 2021-07-30T05:51:13Z

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

@@ -418,14 +419,24 @@ private void ensureSchemaIsLoaded() {
        }
    }

-    private SchemaInfo getSchemaInfo() {
+    public SchemaInfo getSchemaInfo() {
+        if (schemaInfo != null) {


this is quite confusing, we should use what the internal Schema is reporting

codelipenghui · 2021-08-08T05:54:11Z

@315157973 @eolivelli I have addressed your comments, PTAL.

eolivelli

Lgtm

Great work

* [PIP-88] Replicate schemas accross clusters Here is the proposal: https://github.com/apache/pulsar/wiki/PIP-88%3A-Replicate-schemas-across-multiple For the implementation, we just need to set the correct SchemaInfo for the replicated message and using the AutoProduceByte schema for the producer of the

### Motivation apache#11441 supports replicate schema to remote clusters. But there is a mistake that the returned schema state is incorrect. https://github.com/apache/pulsar/blob/e826d849ceef9d6aef28569ad57950bba90dfff1/pulsar-client/src/main/java/org/apache/pulsar/client/impl/MessageImpl.java#L765-L770 Because the replicator used MessageImpl will not have the schema. And this will cause the producer to skip the schema upload. https://github.com/apache/pulsar/blob/e826d849ceef9d6aef28569ad57950bba90dfff1/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java#L2147-L2149 We should remove https://github.com/apache/pulsar/blob/e826d849ceef9d6aef28569ad57950bba90dfff1/pulsar-client/src/main/java/org/apache/pulsar/client/impl/MessageImpl.java#L766-L768 To return the correct schema state. And then we should also provide the correct schema hash. If the message is used by the replicator, the schema hash should be based on the replicator schema. Otherwise, it should use based on the schema of the message. ### Modification - Fixed the incorrect returned schema state - Provide the method for getting schema hash for MessageImpl ### Verification Update the test only to create producer to one cluster. Because if create a producer for other clusters, the producer will upload the schema. This is the reason that why the test can get pass before.

### Motivation #11441 supports replicate schema to remote clusters. But there is a mistake that the returned schema state is incorrect. https://github.com/apache/pulsar/blob/e826d849ceef9d6aef28569ad57950bba90dfff1/pulsar-client/src/main/java/org/apache/pulsar/client/impl/MessageImpl.java#L765-L770 Because the replicator used MessageImpl will not have the schema. And this will cause the producer to skip the schema upload. https://github.com/apache/pulsar/blob/e826d849ceef9d6aef28569ad57950bba90dfff1/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java#L2147-L2149 We should remove https://github.com/apache/pulsar/blob/e826d849ceef9d6aef28569ad57950bba90dfff1/pulsar-client/src/main/java/org/apache/pulsar/client/impl/MessageImpl.java#L766-L768 To return the correct schema state. And then we should also provide the correct schema hash. If the message is used by the replicator, the schema hash should be based on the replicator schema. Otherwise, it should use based on the schema of the message. ### Modification - Fixed the incorrect returned schema state - Provide the method for getting schema hash for MessageImpl

codelipenghui requested a review from congbobo184 July 23, 2021 12:08

codelipenghui self-assigned this Jul 23, 2021

codelipenghui added this to the 2.9.0 milestone Jul 23, 2021

codelipenghui requested review from eolivelli, hangc0276, 315157973, merlimat, sijie and gaoran10 July 23, 2021 12:08

codelipenghui added component/schemaregistry doc-not-needed Your PR changes do not impact docs labels Jul 23, 2021

Fix tests.

be97cd7

315157973 reviewed Jul 28, 2021

View reviewed changes

eolivelli requested changes Jul 30, 2021

View reviewed changes

codelipenghui added 2 commits August 8, 2021 13:39

Merge remote-tracking branch 'apache/master' into penghui/pip-88

3cde4de

Fix comments

4d5bdc1

315157973 approved these changes Aug 8, 2021

View reviewed changes

eolivelli approved these changes Aug 8, 2021

View reviewed changes

eolivelli merged commit a09bd68 into apache:master Aug 8, 2021

codelipenghui deleted the penghui/pip-88 branch August 8, 2021 10:55

codelipenghui mentioned this pull request Feb 14, 2022

Failed to create Consumer for the first time #6576

Closed

codelipenghui restored the penghui/pip-88 branch May 17, 2022 01:21

codelipenghui deleted the penghui/pip-88 branch May 17, 2022 01:30

codelipenghui mentioned this pull request Aug 10, 2022

[fix][broker] Fix schema does not replicate successfully #17049

Merged

1 task

BewareMyPower mentioned this pull request Jun 13, 2023

schema copy in geo relicator #20502

Closed

2 tasks

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[PIP-88] Replicate schemas across clusters #11441

[PIP-88] Replicate schemas across clusters #11441

codelipenghui commented Jul 23, 2021

315157973 Jul 26, 2021 •

edited

Loading

codelipenghui Jul 28, 2021

315157973 Jul 30, 2021

eolivelli Jul 30, 2021

eolivelli left a comment

eolivelli Jul 30, 2021

eolivelli Jul 30, 2021

codelipenghui commented Aug 8, 2021

eolivelli left a comment

[PIP-88] Replicate schemas across clusters #11441

[PIP-88] Replicate schemas across clusters #11441

Conversation

codelipenghui commented Jul 23, 2021

Motivation

Verifying this change

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

315157973 Jul 26, 2021 • edited Loading

Choose a reason for hiding this comment

codelipenghui Jul 28, 2021

Choose a reason for hiding this comment

315157973 Jul 30, 2021

Choose a reason for hiding this comment

eolivelli Jul 30, 2021

Choose a reason for hiding this comment

eolivelli left a comment

Choose a reason for hiding this comment

eolivelli Jul 30, 2021

Choose a reason for hiding this comment

eolivelli Jul 30, 2021

Choose a reason for hiding this comment

codelipenghui commented Aug 8, 2021

eolivelli left a comment

Choose a reason for hiding this comment

315157973 Jul 26, 2021 •

edited

Loading