[PIP-43] Support producer to send msg with different schema by yittg · Pull Request #5165 · apache/pulsar

yittg · 2019-09-10T06:48:47Z

Master Issue: #5141

Motivation

Implement part-1 of PIP-43.

Modifications

New message api to specify message schema explicitly;
Mechanism of registering schema on producing;
Batch message container support to check message schema;
Configuration for seamless introduction of this feature;

Verifying this change

Make sure that the change passes the CI checks.

This change added tests and can be verified as follows in SimpleSchemaTest:

newProducerForMessageSchemaOnTopicWithMultiVersionSchema
newProducerForMessageSchemaOnTopicInitialWithNoSchema
newProducerForMessageSchemaWithBatch

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

Dependencies (does it add or upgrade a dependency): (no)
The public API: (yes)
The schema: (yes)
The default values of configurations: (no)
The wire protocol: (yes)
The rest endpoints: (no)
The admin cli options: (no)
Anything that affects deployment: (no)

Documentation

Does this pull request introduce a new feature? (yes)
If yes, how is the feature documented? (JavaDocs)
If a feature is not documented yet in this PR, please create a followup issue for adding the documentation

Docs in followup issue.

sijie · 2019-09-10T21:26:27Z

/cc @congbobo184 for review as well.

sijie

overall looks good. left two comments, PTAL

sijie · 2019-09-13T15:44:42Z

it seems to me that this method is a bit confusing. T is the original type, no? shouldn't it be a different type?

<V> TypedMessageBuilder<V> newMessage(Schema<V> schema);

@sijie yeah, you are right. I would change the signature in the following PR to make this one easier to review. It would break some more inner method of implementation class.

I'll leave a TODO, or just hide this interface for still not complete

I have hidden this interface and left a TODO.

yittg · 2019-09-17T02:01:15Z

ping @sijie @congbobo184 @jiazhai @codelipenghui, Any more comments?
Thanks

codelipenghui

Looks good to me, just left a few comments

codelipenghui

Overall looks good to me, nice work! @yittg

codelipenghui · 2019-09-18T07:38:53Z

run cpp tests
run java8 tests

yittg · 2019-09-18T15:57:58Z

run cpp tests

sijie · 2019-09-18T16:43:49Z

@yittg I think this should be improved. We should avoid calling sync methods.

A better implementation here is:

if a new schema is needed, flush out all the existing pending requests to the broker. because all the existing pending requests use old schemas that are already registered in broker.

introduce a state for the client (e.g. REGISTERING_SCHEMA). Turn the client into REGISTERING_SCHEMA state. All the requests should be added in the pending queue and wait until the schema is registered in the broker.

after a schema is successfully registered in the broker, flush all the pending requests to the broker.

Yeah, agree with you on executing command asynchronously. It's a better implementation, I will get it out.

First try pushed, PTAL.

yittg · 2019-09-24T05:47:41Z

@sijie Does the new implementation make sense?

congbobo184 · 2019-09-26T02:25:42Z

run Integration Tests

yittg · 2019-09-30T12:52:10Z

@sijie @codelipenghui @congbobo184 May I take up some more of your time to PUSH this PR on.

jiazhai · 2019-10-03T08:37:33Z

run Integration Tests

congbobo184 · 2019-10-04T04:20:40Z

run Integration Tests

jiazhai · 2019-10-07T13:08:19Z

run Integration Tests

sijie · 2019-10-14T02:34:18Z

@yittg sorry for late review. I am looking into it today.

sijie · 2019-10-14T02:39:37Z

-                    if (opSendMsg != null) {
-                        processOpSendMsg(opSendMsg);
-                    }
+                    opSendMsgs = Collections.singletonList(batchMessageContainer.createOpSendMsg());


The original code was there to avoid creating collections for non-batching case. I would prefer not rewriting this logic.

It's ok to leave it as it is.

sijie · 2019-10-14T02:40:12Z

    private void processOpSendMsg(OpSendMsg op) {
        try {
-            batchMessageContainer.clear();
+            if (op.msg != null && isBatchMessagingEnabled()) {


any reason why you are adding this logic here?

op.msg means a single non-batch message. we should flush the current batch if needed.
batchMessageContainer.clear(); was moved into batchMessageAndSend.
Then processOpSendMsg not only for batch message.

Don't we need to do the flush, to keep the order or something else?

sijie · 2019-10-14T06:37:45Z

+            return topic.addSchema(schema);
+        } else {
+            return topic.hasSchema().thenCompose((hasSchema) -> {
+                log.info("[{}] {} configured with schema {}",


Can we keep the origin log statement when refactoring the code?

producer_id not carried in CommandGetOrCreateSchema, actually need it?
Or a -1 as placeholder?

sijie · 2019-10-14T06:41:59Z

        return new TypedMessageBuilderImpl<>(this, schema);
    }

+    // TODO: same parameterized type `T` with producer for the moment,


Create a github issue for this change and add the github issue in this TODO comment? It is a good practice to always track TODO items in github issues.

Should I just edit the master issue, with a checklist, or new a issue for each TODO?

a master issue with a checklist will be good enough.

sijie · 2019-10-25T07:04:48Z

@yittg sorry for late response.

yittg force-pushed the message-explicit-schema branch 5 times, most recently from 081945f to 1e5d71d Compare September 10, 2019 10:06

sijie requested review from codelipenghui and jiazhai September 10, 2019 21:25

sijie assigned yittg Sep 10, 2019

sijie added type/feature The PR added a new feature or issue requested a new feature component/schemaregistry labels Sep 10, 2019

sijie added this to the 2.5.0 milestone Sep 10, 2019

yittg force-pushed the message-explicit-schema branch 3 times, most recently from 87b1eec to e7165ef Compare September 11, 2019 03:16

congbobo184 reviewed Sep 11, 2019

View reviewed changes

Comment thread pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java Outdated

yittg force-pushed the message-explicit-schema branch from e7165ef to 24aedfa Compare September 11, 2019 16:01

sijie reviewed Sep 13, 2019

View reviewed changes

yittg force-pushed the message-explicit-schema branch 2 times, most recently from b61d097 to 814ded0 Compare September 14, 2019 14:31

yittg requested a review from sijie September 16, 2019 05:54

yittg force-pushed the message-explicit-schema branch from 814ded0 to d595c0c Compare September 17, 2019 01:59

codelipenghui reviewed Sep 17, 2019

View reviewed changes

yittg force-pushed the message-explicit-schema branch 3 times, most recently from d3913f4 to 2bf3eef Compare September 17, 2019 07:29

codelipenghui approved these changes Sep 17, 2019

View reviewed changes

sijie requested changes Sep 18, 2019

View reviewed changes

sijie mentioned this pull request Sep 19, 2019

[PIP-44] Separate schema compatibility checker for producer and consumer #5227

Merged

support producer to send msg with different schema

e12e435

yittg force-pushed the message-explicit-schema branch from 2bf3eef to a7d89d8 Compare September 21, 2019 09:43

executing registering schema asynchronously

8d5e8c8

yittg force-pushed the message-explicit-schema branch from a7d89d8 to 8d5e8c8 Compare September 21, 2019 10:20

jiazhai approved these changes Oct 7, 2019

View reviewed changes

codelipenghui requested a review from sijie October 8, 2019 10:37

sijie requested changes Oct 14, 2019

View reviewed changes

sijie approved these changes Oct 25, 2019

View reviewed changes

sijie merged commit 13e35f7 into apache:master Oct 25, 2019

yittg deleted the message-explicit-schema branch October 25, 2019 08:10

BewareMyPower mentioned this pull request Mar 10, 2021

Fix schema not added when subscribe an empty topic without schema #9853

Merged

1 task

Uh oh!

Conversation

yittg commented Sep 10, 2019 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Motivation

Modifications

Verifying this change

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

Documentation

Uh oh!

sijie commented Sep 10, 2019

Uh oh!

Uh oh!

sijie left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

yittg Sep 14, 2019 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

yittg commented Sep 17, 2019

Uh oh!

codelipenghui left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

codelipenghui left a comment

Choose a reason for hiding this comment

Uh oh!

codelipenghui commented Sep 18, 2019

Uh oh!

yittg commented Sep 18, 2019

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

yittg commented Sep 24, 2019

Uh oh!

congbobo184 commented Sep 26, 2019

Uh oh!

yittg commented Sep 30, 2019

Uh oh!

jiazhai commented Oct 3, 2019

Uh oh!

congbobo184 commented Oct 4, 2019

Uh oh!

jiazhai commented Oct 7, 2019

Uh oh!

sijie commented Oct 14, 2019

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

yittg commented Sep 10, 2019 •

edited

Loading

yittg Sep 14, 2019 •

edited

Loading