feat: support multiple schema version for producer and consumer #611

oryx2 · 2021-09-03T09:39:47Z

Motivation

Implement PIP-43.
Support multiple schema version for producer and consumer.

Modifications

add schema cache for producer and consumer
add DisableMultiSchema option for producer

Verifying this change

Make sure that the change passes the CI checks.

(Please pick either of the following options)

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: (yes)
The schema: (yes)
The default values of configurations: (yes)
The wire protocol: (no)

Documentation

Does this pull request introduce a new feature? (no)
If yes, how is the feature documented? (not applicable / docs / GoDocs / not documented)
If a feature is not applicable for documentation, explain why?
If a feature is not documented yet in this PR, please create a followup issue for adding the documentation

cckellogg · 2021-09-03T15:56:47Z

pulsar/consumer_partition.go

+	}
+}
+
+func (s *schemaInfoCache) get(key string) (schema *Schema) {


Let's remove the variable names here and below and return values from the functions. It makes the code more difficult to follow especially below since some returns have values and others do not.

cckellogg · 2021-09-03T15:57:30Z

pulsar/impl_message.go

+		if err != nil {
+			return err
+		}
+		return (*schema).Decode(msg.payLoad, v)


Why is casting needed here?

Thanks review. Because schemaCache store the pointer of schema interface, I have change to schema instance .

cckellogg · 2021-09-03T16:02:04Z

pulsar/producer_partition.go

+	}
+
+	if p.options.DisableMultiSchema {
+		if msg.Schema == nil && p.options.Schema == nil && !msg.Schema.GetSchemaInfo().isSame((p.options.Schema).GetSchemaInfo()) {


Can this be simplified or a comment added. I'm having a difficult time understanding.

chiragbparikh · 2021-10-22T23:25:36Z

Hello @cckellogg and @oryx2.
Can this PR be merged. It is a great feature and we are looking forward to using it. Thanks!

chiragbparikh · 2022-01-19T06:36:28Z

Hello @cckellogg and @oryx2. This is a gentle reminder to please review and merge this PR. As I mentioned above we are eagerly waiting for this feature

lhotari · 2022-01-19T09:17:16Z

@chiragbparikh there are merge conflicts in this PR. please resolve them so that the PR can be reviewed

oryx2 · 2022-01-19T10:11:36Z

@lhotari I just resolve them, please check.

lhotari

Good work @oryx2 !

2 whitespace issues caught my eye, other than that LGTM. I'm not a Golang expert, so could you @zzzming please take a look?

lhotari · 2022-01-19T13:46:03Z

pulsar/consumer_partition.go

@@ -143,10 +144,61 @@ type partitionConsumer struct {
 	dlq         *dlqRouter

 	log log.Logger
-
+  providersMutex       sync.RWMutex


Run go fmt to fix all formatting issues

lhotari · 2022-01-19T13:47:04Z

pulsar/producer.go

+  // Disable multiple Schame Version
+  // Default false


lhotari · 2022-01-19T13:51:45Z

@cckellogg PTAL

zzzming · 2022-01-19T14:00:10Z

pulsar/consumer_partition.go

@@ -143,10 +144,61 @@ type partitionConsumer struct {
 	dlq         *dlqRouter

 	log log.Logger
-
+  providersMutex       sync.RWMutex


Run go fmt to fix all formatting issues

zzzming · 2022-01-19T14:03:30Z

pulsar/consumer_partition.go

+	schema = s.cache[key]
+	s.lock.RUnlock()
+	if schema != nil {
+		return


return schema nil

Or
s.lock.RLock()
schema, ok = s.cache[key]
s.lock.RUnlock()
if ok {
return schema, nil
}

zzzming · 2022-01-19T14:07:52Z

pulsar/consumer_partition.go

+	s.lock.Lock()
+	defer s.lock.Unlock()
+
+	s.cache[schemaVersionHash] = schema


Do we allow schema version overwrite?

You are right, overwrite is not allowed. It is just a private function for SchemaInfoCache.Get which do the overwrite check.

zzzming · 2022-01-19T14:11:20Z

pulsar/consumer_partition.go

+	var properties = make(map[string]string)
+	if pbSchema.Properties != nil {
+		for _, entry := range pbSchema.Properties {
+			properties[*entry.Key] = properties[*entry.Value]


since Key and Value are both declared as *string, do you need check the reference is not nil before assigning them to properties?

I found a function do the same job ConvertToStringMap, but no check nil. May be I should add nil check in ConvertToStringMap function, then invoker it.

zzzming · 2022-01-19T14:36:00Z

pulsar/internal/lookup_service.go

@@ -358,6 +378,9 @@ func (h *httpLookupService) GetTopicsOfNamespace(namespace string, mode GetTopic
 	return topics, nil
 }

+func (h *httpLookupService) GetSchema(topic string, schemaVersion []byte) (schema *pb.Schema, err error) {
+	return nil, errors.New("not support")


Can we make this more descriptive? Maybe an error like this helps to debug.
errors.New("GetSchema is supported by httpLookupService")

Thanks, I will add error message.

zzzming · 2022-01-19T14:45:40Z

pulsar/producer_partition.go

+	}
+
+	if p.options.DisableMultiSchema {
+		if msg.Schema != p.options.Schema {


I am not sure if we want to compare the pointer or the actual value. Shema is an interface. So the current evaluation just make sure they are pointing to the same reference. Do you want to compare the value of Schema?

Yes. Maybe I can compare the value of Schema with this.
msg.Schema != nil && p.options.Schema != nil && msg.Schema.GetSchemaInfo().hash() != p.options.Schema.GetSchemaInfo().hash()

chiragbparikh · 2022-01-27T21:51:46Z

@cckellogg @zzzming can we please review the changes on the PR. Thanks in advance.

chiragbparikh · 2022-04-22T21:43:37Z

@cckellogg @wolfstudy can we please get this MR merged. We are eagerly waiting to use this feature.

chiragbparikh · 2022-05-27T18:51:02Z

@cckellogg can we please get a review on this MR? Appreciate it!

…nto apache-master # Conflicts: # pulsar/consumer_partition.go # pulsar/producer_partition.go

RajatBanerjee · 2022-06-23T20:41:23Z

This feature is something that we are eagerly awaiting.

freeznet · 2022-07-04T04:56:20Z

@oryx2 could you please check the CI errors? thanks.

oryx2 · 2022-07-04T05:38:40Z

@freeznet OK

oryx2 · 2022-07-04T06:20:44Z

@freeznet OK

Sorry,I mad some syntax error when resolve conflicts.Already fixed.Thanks for review.

freeznet · 2022-07-05T02:58:21Z

@oryx2 thanks, please check the CI failures again

pomazanov · 2022-07-05T19:25:30Z

I am curious why this PR is getting so little attention. If there is no need for FORWARD_TRANSITIVE strategy in golang client, what would be a good alternative?

cckellogg reviewed Sep 3, 2021

View reviewed changes

feat: support multiple schema version for producer and consumer

0b1bb11

oryx2 force-pushed the multipleschemas branch from b9cea6a to 0b1bb11 Compare September 13, 2021 07:05

oryx2 requested a review from cckellogg September 22, 2021 06:15

Merge branch 'master' into multipleschemas

fa49071

lhotari reviewed Jan 19, 2022

View reviewed changes

zzzming reviewed Jan 19, 2022

View reviewed changes

Fix format the code and add some description for error

b497930

wolfstudy added this to the 0.9.0 milestone Feb 16, 2022

oryx2 added 2 commits June 16, 2022 14:38

Merge branch 'master' of https://github.com/apache/pulsar-client-go i…

04832f5

…nto apache-master # Conflicts: # pulsar/consumer_partition.go # pulsar/producer_partition.go

Merge branch 'apache-master' into multipleschemas

d66d599

freeznet assigned oryx2 Jul 4, 2022

oryx2 added 2 commits July 4, 2022 13:48

Merge branch 'apache:master' into multipleschemas

5c7392a

fix: syntax error

358b93a

freeznet approved these changes Jul 5, 2022

View reviewed changes

fix lint and CI error

4a387ca

freeznet merged commit ea6eccf into apache:master Jul 7, 2022

Gleiphir2769 mentioned this pull request Nov 1, 2022

[Fix] Fix the Send() stuck caused by callback() not being called #880

Merged

1 task

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

feat: support multiple schema version for producer and consumer #611

feat: support multiple schema version for producer and consumer #611

oryx2 commented Sep 3, 2021

cckellogg Sep 3, 2021

cckellogg Sep 3, 2021

oryx2 Sep 4, 2021

cckellogg Sep 3, 2021

chiragbparikh commented Oct 22, 2021

chiragbparikh commented Jan 19, 2022

lhotari commented Jan 19, 2022

oryx2 commented Jan 19, 2022

lhotari left a comment

lhotari Jan 19, 2022

zzzming Jan 19, 2022

lhotari Jan 19, 2022

lhotari commented Jan 19, 2022

zzzming Jan 19, 2022

zzzming Jan 19, 2022

zzzming Jan 19, 2022

zzzming Jan 19, 2022

oryx2 Jan 20, 2022

zzzming Jan 19, 2022

oryx2 Jan 20, 2022

zzzming Jan 19, 2022

oryx2 Jan 20, 2022

zzzming Jan 19, 2022

oryx2 Jan 20, 2022

chiragbparikh commented Jan 27, 2022

chiragbparikh commented Apr 22, 2022

chiragbparikh commented May 27, 2022

RajatBanerjee commented Jun 23, 2022

freeznet commented Jul 4, 2022

oryx2 commented Jul 4, 2022

oryx2 commented Jul 4, 2022

freeznet commented Jul 5, 2022

pomazanov commented Jul 5, 2022

feat: support multiple schema version for producer and consumer #611

feat: support multiple schema version for producer and consumer #611

Conversation

oryx2 commented Sep 3, 2021

Motivation

Modifications

Verifying this change

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

Documentation

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

chiragbparikh commented Oct 22, 2021

chiragbparikh commented Jan 19, 2022

lhotari commented Jan 19, 2022

oryx2 commented Jan 19, 2022

lhotari 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

lhotari commented Jan 19, 2022

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

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

chiragbparikh commented Jan 27, 2022

chiragbparikh commented Apr 22, 2022

chiragbparikh commented May 27, 2022

RajatBanerjee commented Jun 23, 2022

freeznet commented Jul 4, 2022

oryx2 commented Jul 4, 2022

oryx2 commented Jul 4, 2022

freeznet commented Jul 5, 2022

pomazanov commented Jul 5, 2022