Support for several event types in the same topic #71

pleszczy · 2019-09-06T12:08:39Z

Support for several event types in the same topic

Kafka Avro library will now support several events types in the same topic. This requires using TopicRecordNameStrategy strategy.

const KafkaAvro = require('kafka-avro');

const kafkaAvro = new KafkaAvro({
    kafkaBroker: 'localhost:9092',
    schemaRegistry: 'http://localhost:8081',
    keySubjectStrategy: "TopicRecordNameStrategy",
    valueSubjectStrategy: "TopicRecordNameStrategy",
});

You can read more about this here : https://www.confluent.io/blog/put-several-event-types-kafka-topic/

…o use type-topicName as a subject for avro schema

…ng multiple types on one topic.

* Using let/const instead of var * Standardizing logging message format * Using template strings * Using standard nodejs err keyword * Reformatting code with default intellij formatter for js

* Using let/const instead of var * Updating circleCi to 2.0

…ng to kafka and schema-registry)

test/spec/producer.test.js

ricardohbin

Wow, great job. But you do a lot of things, instead of only add the strategies:

circle2 (thanks!)
replace dated var to const, template strings, typos etc. This "replace" itself - including add async/await and exampels in readme.md - is a "feature" that demands one minor release.
add the TopicRecordNameStrategy strategy

All this could be 3 PR's - it's very hard to review :/ BTW, you are replacing completely the current TopicNameStrategy by ? (or I miss some point?). If you do, we need to support both strategies since TopicNameStrategy is the default.

And other few comments.

What do you think?

test/spec/magic-byte.test.js

lib/kafka-consumer.js

ricardohbin · 2019-09-06T17:56:12Z

lib/kafka-consumer.js

@@ -101,65 +99,60 @@ Consumer.prototype._onWrapper = function (consumerInstance, eventName, cb) {
    return consumerInstance.__kafkaAvro_on(eventName, cb);
  }

-  return consumerInstance.__kafkaAvro_on('data', function(message) {
-    if (!this.sr.keySchemas[message.topic]) {


why you remove this guard clause?

That's because I've moved away from this mechanism in favor of extracting schemaId from the magic byte. There is the same warning being logged a couple of lines below though.

ricardohbin · 2019-09-06T17:57:40Z

circle.yml

+            nvm alias default v8.16
+            # Each step uses the same `$BASH_ENV`, so need to modify it
+            echo 'export NVM_DIR="/opt/circleci/.nvm"' >> $BASH_ENV
+            echo "[ -s \"$NVM_DIR/nvm.sh\" ] && . \"$NVM_DIR/nvm.sh\"" >> $BASH_ENV


we can simplify this if we don't use nvm - just raw node. What you think?

We need v8 version of node to build node-rdkafka. Ill see if I can use apt-get to install a concrete version of node.

I tried but I couldn't get apt-get to install the version I wanted. I've moved back to nvm.

command: | curl -sL https://deb.nodesource.com/setup_8.x | sudo -E bash - sudo apt install nodejs node --version

`

https://circleci.com/gh/pleszczy/kafka-avro/63

README.md

ricardohbin · 2019-09-06T18:00:25Z

lib/kafka-producer.js

-    ? this.sr.valueSchemas[topicName]
-    : this.sr.keySchemas[topicName];
+  // implementing io.confluent.kafka.serializers.subject.TopicRecordNameStrategy
+  // FIXME: It will break if code gets minimized


why the code would be minimized?

What I meant is if you use this library in your project and minimize the project and its dependencies. It'll break because the the type name will be converted to a random string.

ricardohbin · 2019-09-06T18:01:23Z

lib/magic-byte.js

-    throw new TypeError('Message not serialized with magic byte');
+magicByte.fromMessageBuffer = function (encodedMessage, sr) {
+  if (!encodedMessage || encodedMessage[0] !== MAGIC_BYTE) {
+    return null;


why are you returning null instead of throw TypeError?

That because it's simpler to me to just check for null instead of doing a try catch block. But it doesn't matter that much to me and it's up to you preferred code style. If you would like I can convert it back to try-catch.

yep, it is simple, but the entire code is based in try-catch, let's try to break less possible

ricardohbin · 2019-09-06T18:02:48Z

lib/magic-byte.js

-  } else {
-    decoded = sr.schemaTypeById[schemaKey].decode(encodedMessage, 5);
+  if (!schemaType) {
+    return null;


this shouldnt be return type.decode(encodedMessage, 5); to keep the previous behavior?

I have changed the behaviour here. It was unnecessary since all we need is just the magic byte and sr.schemaTypeById map.

pleszczy

Wow, great job. But you do a lot of things, instead of only add the strategies:

Apologies for that, I needed this functionality and I usually do refactoring while working on something too. I have also needed a working CI to keep track of builds. I did hold off on converting to TypeScript though.

circle2 (thanks!)

replace dated var to const, template strings, typos etc. This "replace" itself - including add async/await and exampels in readme.md - is a "feature" that demands one minor release.

add the TopicRecordNameStrategy strategy

All this could be 3 PR's - it's very hard to review :/ BTW, you are replacing completely the current TopicNameStrategy by ? (or I miss some point?). If you do, we need to support both strategies since TopicNameStrategy is the default.

You are right here, I'll implement all the confluent strategies.

And other few comments.

What do you think?

Thank you for the constructive review. I have implemented most of the comments and I'll implement the other confluent strategies now.

pleszczy · 2019-09-09T07:00:46Z

lib/kafka-consumer.js

@@ -101,65 +99,60 @@ Consumer.prototype._onWrapper = function (consumerInstance, eventName, cb) {
    return consumerInstance.__kafkaAvro_on(eventName, cb);
  }

-  return consumerInstance.__kafkaAvro_on('data', function(message) {
-    if (!this.sr.keySchemas[message.topic]) {


That's because I've moved away from this mechanism in favor of extracting schemaId from the magic byte. There is the same warning being logged a couple of lines below though.

pleszczy · 2019-09-09T07:07:35Z

lib/kafka-producer.js

-    ? this.sr.valueSchemas[topicName]
-    : this.sr.keySchemas[topicName];
+  // implementing io.confluent.kafka.serializers.subject.TopicRecordNameStrategy
+  // FIXME: It will break if code gets minimized


What I meant is if you use this library in your project and minimize the project and its dependencies. It'll break because the the type name will be converted to a random string.

pleszczy · 2019-09-09T07:09:57Z

circle.yml

+            nvm alias default v8.16
+            # Each step uses the same `$BASH_ENV`, so need to modify it
+            echo 'export NVM_DIR="/opt/circleci/.nvm"' >> $BASH_ENV
+            echo "[ -s \"$NVM_DIR/nvm.sh\" ] && . \"$NVM_DIR/nvm.sh\"" >> $BASH_ENV


We need v8 version of node to build node-rdkafka. Ill see if I can use apt-get to install a concrete version of node.

pleszczy · 2019-09-09T07:15:06Z

lib/magic-byte.js

-    throw new TypeError('Message not serialized with magic byte');
+magicByte.fromMessageBuffer = function (encodedMessage, sr) {
+  if (!encodedMessage || encodedMessage[0] !== MAGIC_BYTE) {
+    return null;


That because it's simpler to me to just check for null instead of doing a try catch block. But it doesn't matter that much to me and it's up to you preferred code style. If you would like I can convert it back to try-catch.

pleszczy · 2019-09-09T07:19:26Z

lib/magic-byte.js

-  } else {
-    decoded = sr.schemaTypeById[schemaKey].decode(encodedMessage, 5);
+  if (!schemaType) {
+    return null;


I have changed the behaviour here. It was unnecessary since all we need is just the magic byte and sr.schemaTypeById map.

…m to apt-get

… back to nvm.

ricardohbin

Hmm thanks for the adjusts. I just replied some comments

About Typescript, there is an issue that can be related with it (#59) - but it would be a complete refactoring, almost a full rewrite. I don't see the advantages of doing this here :/

I dumped 2.0.0 with lidbrdkafka today, please update your branch and see everything is ok. Then, we can test your version with real scenarios before the merge of possible 3.0.0

ricardohbin · 2019-09-09T20:13:44Z

lib/kafka-consumer.js

@@ -101,65 +99,60 @@ Consumer.prototype._onWrapper = function (consumerInstance, eventName, cb) {
    return consumerInstance.__kafkaAvro_on(eventName, cb);
  }

-  return consumerInstance.__kafkaAvro_on('data', function(message) {
-    if (!this.sr.keySchemas[message.topic]) {


ricardohbin · 2019-09-09T20:14:44Z

lib/magic-byte.js

-    throw new TypeError('Message not serialized with magic byte');
+magicByte.fromMessageBuffer = function (encodedMessage, sr) {
+  if (!encodedMessage || encodedMessage[0] !== MAGIC_BYTE) {
+    return null;


yep, it is simple, but the entire code is based in try-catch, let's try to break less possible

lib/subject-strategy.js

pleszczy · 2019-09-10T07:57:47Z

Hmm thanks for the adjusts. I just replied some comments

About Typescript, there is an issue that can be related with it (#59) - but it would be a complete refactoring, almost a full rewrite. I don't see the advantages of doing this here :/

I dumped 2.0.0 with lidbrdkafka today, please update your branch and see everything is ok. Then, we can test your version with real scenarios before the merge of possible 3.0.0

Done.

lib/kafka-consumer.js

[Node.js dependencies (e.g. require('util'))] [Third Party dependencies (e.g. require('axios'))] [Local dependencies (e.g. require('./local'))]

pleszczy · 2019-09-16T10:51:51Z

@ricardohbin I did local testing and it was successfully. Hope this can be merged to upstream.

ricardohbin · 2019-09-17T18:25:16Z

@pleszczy nice! I will just do a double check with some large scale production data to see everything is ok - if everything goes ok I will ship in this week.

Can you add yourself with this feature description in README CHANGELOG plz? It would be 3.0.0 probably (in concept it could be 2.1.0, but how this PR add a new feature + whole feature, it's better go with caution and avoid breaks)

pleszczy · 2019-09-18T07:43:15Z

@pleszczy nice! I will just do a double check with some large scale production data to see everything is ok - if everything goes ok I will ship in this week.

Can you add yourself with this feature description in README CHANGELOG plz? It would be 3.0.0 probably (in concept it could be 2.1.0, but how this PR add a new feature + whole feature, it's better go with caution and avoid breaks)

Done, thank you.

ricardohbin · 2019-09-24T12:55:39Z

@pleszczy I tested here and everything is ok in my test. Thank you very much for this big contribution (feature + code upgrade + CI/CD).

Shipping 3.0.0!

Piotr Leszczynski added 30 commits August 29, 2019 10:20

Chore(Support multiply types in a single topics): Changing producer t…

dec045b

…o use type-topicName as a subject for avro schema

Chore(Support multiply types in a single topics): Reading and produci…

02682d0

…ng multiple types on one topic.

Chore(Support multiply types in a single topics): Fixing the base tests.

7597194

refactoring(Support multiply types in a single topics):

2f65ca4

* Using let/const instead of var * Standardizing logging message format * Using template strings * Using standard nodejs err keyword * Reformatting code with default intellij formatter for js

refactoring(Support multiply types in a single topics):

8db6159

* Using let/const instead of var * Updating circleCi to 2.0

chor(Support multiply types in a single topics): Enable circleCi

2e4fd93

chor(Support multiply types in a single topics): Enable circleCi

7204e4c

chor(Support multiply types in a single topics): Enable circleCi

a18d644

chor(Support multiply types in a single topics): Enable circleCi

70992bc

chore(Setting up circleCi)

de4a90a

chore(Setting up circleCi)

e58d81b

chore(Setting up circleCi)

3102dc6

chore(Setting up circleCi)

920b5d5

Enable docker

fe779dc

Enable docker

807455f

configuration(Setting up tests)

3d4a4fa

configuration(Setting up tests)

a6b110c

configuration(downgrading npm version used by circleCi)

af33bd5

configuration(downgrading npm and node versions used by circleCi)

e578795

configuration(downgrading npm and node versions used by circleCi)

25a197b

configuration(downgrading npm and node versions used by circleCi)

51930e5

configuration(downgrading npm and node versions used by circleCi)

26680e0

configuration(Setting up tests)

3b41466

configuration(Setting up tests)

aa86743

configuration(Setting up tests)

ab2c06d

configuration(merging build and test workflows into one in cricleCi)

4ac4bc5

configuration(setting up kafka hostnames in cricleCi)

22a8f5f

configuration(setting up kafka hostnames in cricleCi)

b75c9e0

configuration(cricleCi)

2fd0801

configuration(troubleshooting circleCi network problems when connecti…

d72e243

…ng to kafka and schema-registry)

documentation()

a1d6066

moskalap reviewed Sep 6, 2019

View reviewed changes

test/spec/producer.test.js Outdated Show resolved Hide resolved

moskalap approved these changes Sep 6, 2019

View reviewed changes

pr(fixing a typo)

adb1b0c

ricardohbin reviewed Sep 6, 2019

View reviewed changes

pleszczy commented Sep 9, 2019

View reviewed changes

Piotr Leszczynski and others added 10 commits September 9, 2019 09:26

pr(implementing comments)

e2612d2

Merge branch 'master' into master

83c2eec

pr(implementing comments)

996327a

Merge remote-tracking branch 'origin/master'

aca5fa7

Changing how nodejs is being installed on circleCi. Switching from nv…

b3a984c

…m to apt-get

Changing how nodejs is being installed on circleCi. Switching from nv…

1c0d969

…m to apt-get

Changing how nodejs is being installed on circleCi. Switching from nv…

eaa616f

…m to apt-get

Changing how nodejs is being installed on circleCi. Switching from nv…

75b8ca8

…m to apt-get

Reverting : Changing how nodejs is being installed on circleCi. Going…

7887800

… back to nvm.

Updating documentation.

2788b5d

ricardohbin reviewed Sep 9, 2019

View reviewed changes

Implementing pull request comments.

0c042e1

thanpolas reviewed Sep 10, 2019

View reviewed changes

lib/kafka-consumer.js Show resolved Hide resolved

Formatting requierments according to :

cff58a2

[Node.js dependencies (e.g. require('util'))] [Third Party dependencies (e.g. require('axios'))] [Local dependencies (e.g. require('./local'))]

Updating changelog.

d33c352

Merge branch 'master' into master

e7b882f

ricardohbin merged commit cfc54d6 into waldophotos:master Sep 24, 2019

ricardohbin mentioned this pull request Sep 24, 2019

Update circleCI to 2.0 #46

Closed

ricardohbin mentioned this pull request Apr 27, 2020

Adding support to specifiy __schemaName/__id to use for avro serialization. #91

Closed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Support for several event types in the same topic #71

Support for several event types in the same topic #71

pleszczy commented Sep 6, 2019 •

edited

Loading

ricardohbin left a comment

ricardohbin Sep 6, 2019

pleszczy Sep 9, 2019 •

edited

Loading

ricardohbin Sep 9, 2019

ricardohbin Sep 6, 2019

pleszczy Sep 9, 2019

pleszczy Sep 9, 2019 •

edited

Loading

ricardohbin Sep 6, 2019

pleszczy Sep 9, 2019

ricardohbin Sep 6, 2019

pleszczy Sep 9, 2019

ricardohbin Sep 9, 2019

ricardohbin Sep 6, 2019

pleszczy Sep 9, 2019

pleszczy left a comment •

edited

Loading

pleszczy Sep 9, 2019 •

edited

Loading

pleszczy Sep 9, 2019

pleszczy Sep 9, 2019

pleszczy Sep 9, 2019

pleszczy Sep 9, 2019

ricardohbin left a comment

ricardohbin Sep 9, 2019

ricardohbin Sep 9, 2019

pleszczy commented Sep 10, 2019

pleszczy commented Sep 16, 2019 •

edited

Loading

ricardohbin commented Sep 17, 2019

pleszczy commented Sep 18, 2019

ricardohbin commented Sep 24, 2019

Support for several event types in the same topic #71

Support for several event types in the same topic #71

Conversation

pleszczy commented Sep 6, 2019 • edited Loading