Fix rpc exception kafka #10379
Fix rpc exception kafka #10379
Conversation
try fix from nestjs#10113 (comment)
Bug caused microservise not to return error to Kafka transport when using custom RPC exception filter Fixes issue nestjs#10113
Pull Request Test Coverage Report for Build d8d9a5ab-12b5-4810-9ec6-34797051705d
💛 - Coveralls |
|
Hey @kamilmysliwiec could you take a quick look at this PR? It seems to fix an issue introduced by this commit: cfe1191 |
| if (!isPromiseResolved) { | ||
| isPromiseResolved = true; | ||
| resolve(); | ||
| } |
There was a problem hiding this comment.
Why this condition is removed? What if a stream emits multiple values (streaming)?
There was a problem hiding this comment.
Maybe @kimvladis can help with this?
Credits for this commit belongs to @kimvladis because he suggested it in #10113 (comment)
I just checked for his solution and it worked for me.
There was a problem hiding this comment.
Why this condition is removed? What if a stream emits multiple values (streaming)?
Reasoned, but as I can see, there are no supporting for multiple values emitting in handlers.
I think this condition can be returned, just in case. @RaiMX
There was a problem hiding this comment.
@kimvladis emitting multiple values is supported
There was a problem hiding this comment.
Why this condition is removed? What if a stream emits multiple values (streaming)?
Reasoned, but as I can see, there are no supporting for multiple values emitting in handlers. I think this condition can be returned, just in case. @RaiMX
Honestly I can't seem to figure this out. Can you take it from here @kimvladis? Or maybe suggest edit, then I will merge it?
There was a problem hiding this comment.
Sorry, I think I realised scenario, but we never used it like that in our code. So the handler can return observable, and the client can receive multiple values, and multiple values will be emitted to Kafka(same topic, same correlationId)?
There was a problem hiding this comment.
By the way, this condition is redundant, because resolving already resolved promises has no effects. Or I missed something?
https://tc39.es/ecma262/#sec-promise-objects
|
So no updates at the moment? |
|
I was about to open a PR to add that |
|
@kamilmysliwiec no updates yet? |
|
+1 Please someone look into it. @kamilmysliwiec |
|
I added the following code to my application as a workaround for now: // created file kafka-server.ts
// HACK - This is a hack to fix the issue with the @nestjs/microservices package tracked in https://github.com/nestjs/nest/pull/10379
// eslint-disable-next-line @typescript-eslint/ban-ts-comment
// @ts-ignore
export class KafkaServer extends ServerKafka {
private combineStreamsAndThrowIfRetriable(
response$: { subscribe: (arg0: { next: (val: any) => void; error: (err: any) => void; complete: () => any }) => void },
replayStream$: { next: (arg0: any) => void; error: (arg0: any) => void; complete: () => any },
) {
return new Promise<void>((resolve, reject) => {
let isPromiseResolved = false;
response$.subscribe({
next: val => {
replayStream$.next(val);
if (!isPromiseResolved) {
isPromiseResolved = true;
resolve();
}
},
error: err => {
if (err instanceof KafkaRetriableException && !isPromiseResolved) {
isPromiseResolved = true;
reject(err);
}
replayStream$.error(err);
resolve(); // This line is needed to prevent server stall
},
complete: () => replayStream$.complete(),
});
});
}
}// on main.ts
app.connectMicroservice<MicroserviceOptions>(
{
strategy: new KafkaServer({
client: {
brokers: environment.KAFKA_URL.map(url => `${url}:${environment.KAFKA_PORT}`),
clientId: 'logs',
},
consumer: { groupId: 'logs' },
}),
},
{ inheritAppConfig: true },
);
await app.startAllMicroservices(); |
|
Fixed in this PR #10982 |
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?
#10113
Issue Number: 10113
What is the new behavior?
Microservice return custom RPC error through Kafka transport
Does this PR introduce a breaking change?
Other information