feat(microservice) Kafka Support #2361

[mkaufmaner]

PR Checklist

Please check if your PR fulfills the following requirements:

• The commit message follows our guidelines: https://github.com/nestjs/nest/blob/master/CONTRIBUTING.md
• Tests for the changes have been added (for bug fixes / features)
• Docs have been added / updated (for bug fixes / features)

PR Type

What kind of change does this PR introduce?

[ ] Bugfix
[x] Feature
[ ] Code style update (formatting, local variables)
[ ] Refactoring (no functional changes, no api changes)
[ ] Build related changes
[ ] CI related changes
[ ] Other... Please describe:

What is the current behavior?

#2361

Issue Number: #2361

What is the new behavior?

Does this PR introduce a breaking change?

[ ] Yes
[x] No

Other information

[mkaufmaner]

Whoops, looks like I messed up some dependencies. Fixing now.

[mkaufmaner]

@kamilmysliwiec I have no illusions that this is complete. However, your input would be greatly appreciated.

[kamilmysliwiec]

Can you elaborate on how the communication is being handled right now?

[mkaufmaner]

@kamilmysliwiec would love to have this included in 6.6.0. Let me know what I have to do to make that happen and I will.

[Hexenon]

@mkaufmaner is it ready?, do you need any other help?, i hope my help was enough.

can't wait to use it, i need this Kafka message queue for my application.

Also, i think we need to add a how to use this to docs repo

[kamilmysliwiec]

We still have to cover these 2 things:

• integration with serializers/deserializers feature feat(microservices): add serializer and deserializer, improve performance #2653
• PR to the documentation with a new transport layer under Microservices

[mkaufmaner]

@kamilmysliwiec Alright, I'll merge 6.6.0 into this branch, integrate the serializer/deserializers and work on the documentation.

[mkaufmaner]

@kamilmysliwiec Should I extend the the serializer/deserializers with Kafka specific serializers (ie. incoming-request-kafka.deserializer.ts)? Or should I implement the parsing of kafka message elsewhere?

[kamilmysliwiec]

The key responsibility of serializers/deserializers is to adjust incoming/outgoing packages to the unique format required by NestJS. In the best-case scenario, you should use the exact same serializers for Kafka as we use for any other strategy (since Kafka should be considered as an official strategy eventually)

[mkaufmaner]

The key responsibility of serializers/deserializers is to adjust incoming/outgoing packages to the unique format required by NestJS. In the best-case scenario, you should use the exact same serializers for Kafka as we use for any other strategy (since Kafka should be considered as an official strategy eventually)

Understandable. However, I don't really see any other way. There is the problem where message keys, values, and headers are buffers which would require the developer to implement their own serialization within their controllers. Subsequently, when a developer changes transports, they would have to modify their controllers significantly. There is also a problem maintaining interoperability within the cluster if we utilize the existing serializers/deserializers. This is because message headers are utilized for message metadata like correlation ids, reply topics, and reply partitions.

[kamilmysliwiec]

@mkaufmaner unfortunately, I can't tell you more before I dive into the code :( I'll try to do review it as soon as possible

[mkaufmaner]

@kamilmysliwiec I see that in some servers like server-mqtt.ts and server-redis.ts you have a method for actually parsing the raw message (parseMessage). I would think this would be handled by the serializer. Is this correct?

[kamilmysliwiec]

@mkaufmaner actually, no. That's the reason why transport-related parsing is located directly in classes, whereas high-level serialization/deserialization (alignment to external/internal systems) is being done in dedicated serializers/deserializers. Hence, they can be shared among different transport strategies by default.

[mkaufmaner]

@mkaufmaner actually, no. That's the reason why transport-related parsing is located directly in classes, whereas high-level serialization/deserialization (alignment to external/internal systems) is being done in dedicated serializers/deserializers. Hence, they can be shared among different transport strategies by default.

@kamilmysliwiec Gotcha.

[mkaufmaner]

@kamilmysliwiec I implemented the serializers/deserializers. However, can't seem to get the CircleCI builds to run.

[kamilmysliwiec]

Thank you @mkaufmaner. I have adjusted performed changes to the current codebase style and merged PR. I hope to publish this feature within 6.7.0 release.

[wejson]

I c this is released, Thank u :)

Is there any documentation? @mkaufmaner
i really looking forward to using this with the cqrs

[vjtc0n]

@kamilmysliwiec Hello, Great work!
But it's November now, can we have an example to work with?
Thanks.

[lock]

This thread has been automatically locked since there has not been any recent activity after it was closed. Please open a new issue for related bugs.