fix(microservices): messages emitted before kafka client is ready fail #11026
Conversation
| @@ -94,46 +94,56 @@ export class ClientKafka extends ClientProxy { | |||
| this.consumer && (await this.consumer.disconnect()); | |||
| this.producer = null; | |||
| this.consumer = null; | |||
| this.initialized$ = null; | |||
There was a problem hiding this comment.
@kamilmysliwiec is there any convention defined in Nestjs for the use of the $ suffix?
I'm aware that is an Angular convention to define Observable but doesn't seem to express any special type here
There was a problem hiding this comment.
This might be somewhat confusing indeed @overbit. We usually use $ for observables, we don't have any specific convention dedicated to promises. Let me fix this up
Pull Request Test Coverage Report for Build 9ca3f05f-9a28-4b4b-9c3a-9d826bebd6fa
💛 - Coveralls |
delete-merged-branch
bot
deleted the
fix/kafka-client-parallel-messages
branch
| this.client = null; | ||
| } | ||
|
|
||
| public async connect(): Promise<Producer> { | ||
| if (this.client) { | ||
| return this.producer; | ||
| return this.initialized$.then(() => this.producer); |
There was a problem hiding this comment.
Another question around consistency:
in the rest of the function, we use async await but in there we leverage .then.
I'd suggest keeping the same approach here to make the code more readable even if the execution is not a 100% replacement of the promise chaining.
| return this.initialized$.then(() => this.producer); | |
| await this.initialized$; | |
| return this.producer; |
There was a problem hiding this comment.
.then is intentionally used here to indicate the intent. The Promise here is used for slightly different purpose than other promises in this class
| groupId: this.groupId, | ||
| }, | ||
| ); | ||
| this.initialized$ = new Promise(async (resolve, reject) => { |
There was a problem hiding this comment.
Would it make more sense to split the initialized promise into two promises consumer and producer?
Conceptually, a service can start producing without the need to wait for a Kafka consumer connection to be established.
In Event-driven architecture, we need to aim for produce as soon as required but consume only when the service is ready to accept.
@kamilmysliwiec what do you think?
There was a problem hiding this comment.
Actually we could just create a promise for the consumer.connect and consumer.bind and not await for it so the function connect can continue to execute and await only for producer.connect.
I'll draft a PR with the idea to make it more clear
There was a problem hiding this comment.
Conceptually, a service can start producing without the need to wait for a Kafka consumer connection to be established.
Well, it depends. If you're (for some reason) using the message-based approach then you should wait for a Kafka consumer connection to be established before you start producing messages.
If you don't need to consume messages on the other hand, you could just use the producerOnly mode which skips the initialization of the consumer - in this case we don't need 2 promises either.
So IMO we should leave it as is
There was a problem hiding this comment.
Good point. That is unfortunately to force the same approach for both but it would require a major refactor so I get it.
Thanks @kamilmysliwiec
|
#11029 @kamilmysliwiec if that makes sense I'll properly fix the PR and we can revert this PR |
PR Checklist
Please check if your PR fulfills the following requirements:
PR Type
What kind of change does this PR introduce?
What is the current behavior?
Issue Number: #10449
What is the new behavior?
Does this PR introduce a breaking change?
Other information