feat(microservices): errors thrown during eachMessage execution should be passed to kafkajs

[davidschuette]

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
• 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?

Issue Number: #9283

What is the new behavior?

Errors thrown during kafka event processing are no longer being caught by the RcpProxy but passed to kafkajs's eachMessage.
KafkaContext provides function to get reference to consumer instance.

Does this PR introduce a breaking change?

• Yes
• No

If the previous functionality is required a custom interceptor can be used to catch the errors.

Other information

I did not add any tests for the reworked error handling. I will add them if the changes get approved.

[coveralls]

Pull Request Test Coverage Report for Build a6202d37-e0ab-4c77-8747-24db9117807b

• 93 of 103 (90.29%) changed or added relevant lines in 20 files are covered.
• 72 unchanged lines in 8 files lost coverage.
• Overall coverage increased (+0.1%) to 94.17%

---

Changes Missing Coverage	Covered Lines	Changed/Added Lines	%
packages/microservices/client/client-tcp.ts	4	5	80.0%
packages/microservices/server/server-tcp.ts	2	3	66.67%
packages/microservices/server/server.ts	0	1	0.0%
packages/core/router/router-explorer.ts	9	11	81.82%
packages/microservices/server/server-mqtt.ts	0	2	0.0%
packages/microservices/helpers/tcp-socket.ts	29	32	90.63%

Files with Coverage Reduction	New Missed Lines	%
packages/microservices/client/client-kafka.ts	3	95.63%
packages/microservices/server/server-tcp.ts	4	76.74%
packages/core/router/router-explorer.ts	5	82.76%
packages/microservices/server/server-mqtt.ts	5	81.34%
packages/microservices/client/client-rmq.ts	6	84.87%
packages/microservices/client/client-tcp.ts	6	86.84%
packages/core/injector/module.ts	9	84.58%
packages/core/injector/injector.ts	34	74.13%

Totals
Change from base Build 0a7dce2a-d55f-4ed3-96fe-535d2466a616:	0.1%
Covered Lines:	5767
Relevant Lines:	6124

---

💛 - Coveralls

[davidschuette]

@kamilmysliwiec Any thoughts on the new handling of errors in eachMessage? If this is approved I will add tests and update my PR for the docs.

[kamilmysliwiec]

@davidschuette yeah this change makes a lot of sense, agree.

[davidschuette]

@kamilmysliwiec I fixed the merge conflict. Is there anything else I need to do to get this PR approved?

[kamilmysliwiec]

PR should be all set @davidschuette. I'll review it as soon as I can

[mastermatt]

I'm a fan of this PR, and I'd like to see it released ASAP. Unfortunately, I don't think it's enough to solve some scenarios. I've been hunting down an issue with how ServerKafka interacts with kafkjs's eachMessage when I found this PR. If my issue ends up being unrelated, I'll open a separate bug issue, but I'm pretty sure they're intertwined and the current change here is only a half-fix.
What I discovered was an inconsistency in what gets returned to eachMessage. What should happen, in all cases, is ServerKafka.handleMessage should return a Promise that resolves or rejects after awaiting all the user-land code in the controller and any interceptors etc. However, in the case of event handlers where an Observable is introduced into the response flow, the Promise returned to eachMessage resolves before any user-land code is run.
I believe the issue should be fixed here:

nest/packages/microservices/server/server.ts

Lines 118 to 125 in 26e9ed6

	const resultOrStream = await handler(packet.data, context);
	if (isObservable(resultOrStream)) {
	const connectableSource = connectable(resultOrStream, {
	connector: () => new Subject(),
	resetOnDisconnect: false,
	});
	connectableSource.connect();
	}

If handler returns a Promise, the await on line 118 will cause the resulting Promise of handleEvent async-ness to correctly wait and allow any errors to bubble up the stack. However, if an Observable is resolved from handler then handleEvent will return a Promise that resolves too quickly and never rejects.
Thanks to this little bit of code, simply adding Interceptors will cause any value returned from the controller to be wrapped in an Observable and cause this bug.

I think the solution is to add await lastValueFrom(connectableSource); after line 124 of Server.handleEvent, but my RxJS-foo is weak and could use a second opinion.

[mastermatt]

In the case of a CustomStrategy, the transport needs to come from config.strategy.transportId

[davidschuette]

@mastermatt Thank you for your support on this PR.
I changed a few lines to fix the problem you described (see a09572e).
I don't know what git did with all these other commits showing up now... any help is appreciated.

[mastermatt]

I'm not quite following the expected behavior of removing the exceptionsHandler from the RPC Proxy logic.
@davidschuette can you clarify?
When I play around with this locally, the errors propagate to kafjajs' eachMessage by simplifying adding the await lastValueFrom(resultOrStream); line. However, when I add the other changes (KafkaRpcProxy) the error still makes it to kafkajs, but none of my custom exception filters fire.

And one other question, sorry for hijacking this thread, what should be returned from a filter implementing RpcExceptionFilter.catch to stop the error propagation? Or should an interceptor be used for that?

[davidschuette]

@mastermatt It looks like your fix is way easier than mine. I don't know how I missed that.

Do you know what this code is doing? I removed it and it doesn't break any tests.
Line 120 in packages/microservices/server/server.ts

const connectableSource = connectable(resultOrStream, {
  connector: () => new Subject(),
  resetOnDisconnect: false,
});
connectableSource.connect();

[davidschuette]

@mastermatt Regarding your issue with filters. I will take a look at that problem tomorrow.

[mastermatt]

@davidschuette keeping in mind I don't have much RxJS experience, I think the connectableSource bit is there to kick off the observable if it's deferred. But since lastValueFrom calls subscribe, I think the connectableSource becomes redundant.
My local testing was also successful with only the changes from a09572e

[davidschuette]

@mastermatt Can you give me a repo with the project that is giving you problems with filters?

[kamilmysliwiec]

Btw @davidschuette the commit history of this PR got all messed up by pulling changes from master (instead of simply merging that branch). Can you rebase your changes?

[davidschuette]

@kamilmysliwiec I will give it a try. Though I don't have much experience with rebasing.
The branch got messed up as a result of a rebase.

[gkampitakis]

Any updates on this ? Could we help somehow to have this pr merged 🤔 ?

[mastermatt]

Ignoring the current state of the rebase, let me take a stab at trying to summarize the current state of this change set:

a. Update Server#handleEvent to ensure any Observable is finished before resolving the async method.
a09572e
This change isn't error-handling specific, instead it just allows a Kafka message handler to either resolve or reject before kafkajs moves on to the next message in the batch.
For my own personal needs, this change did the trick, and I've been running a version of this in Production successfully for a few weeks now.

b. Add consumer getter to Kafka context.
b91803c
A nice feature add that allows consumers to manually interact with the kafkajs Consumer instance when some non-generic action is needed.

c. Add KafkaRpcProxy that bypasses error handling after a controller method throws an error (skips the Exception filters)
77b7bd6
This is the part I'm unsure about, because it's not clear from the docs what the expected behavior is in Nest.
@kamilmysliwiec can you clarify: what the expected (desired) behavior of Exception Filters for RPC event-style messages? In my experience, the only things that could happen, if the Controller throws, is to determine if/how the error should cause a retry, be ignored, or deliver to a dead-letter-queue.

[gkampitakis]

Really nice sum. Would it make sense to unblock this pr by splitting it into the prs that are more trivial and sure that are correct?

[mastermatt]

I would be happy to open a pr or two, if that would help, but at this point I don't think there is a blocker beyond simply not knowing the desired behavior. We need clarity on the intended design/architecture.

[kamilmysliwiec]

we're tracking this here #9586