feat(microservices): dont catch errors thrown by kafka event handling

Error thrown by event handling method are no longer being caught by RcpExecptionFilter.
Instead errors are passed to kafkajs's eachMessage.
This results in proper interaction with kafka.

This commit closes Kafka commitOffsets #9283.

packages/microservices/context/kafka-rpc-proxy.ts

@@ -0,0 +1,14 @@
+import { RpcExceptionsHandler } from 'exceptions/rpc-exceptions-handler';
+import { Observable } from 'rxjs';
+import { RpcProxy } from './rpc-proxy';
+
+export class KafkaRpcProxy extends RpcProxy {
+  public create(
+    targetCallback: (...args: unknown[]) => Promise<Observable<any>>,
+    exceptionsHandler: RpcExceptionsHandler,
+  ): (...args: unknown[]) => Promise<Observable<unknown>> {
+    return (...args: unknown[]) => {
+      return targetCallback(...args);
+    };
+  }
+}

packages/microservices/microservices-module.ts

@@ -13,8 +13,10 @@ import { PipesContextCreator } from '@nestjs/core/pipes/pipes-context-creator';
 import { ClientProxyFactory } from './client';
 import { ClientsContainer } from './container';
 import { ExceptionFiltersContext } from './context/exception-filters-context';
+import { KafkaRpcProxy } from './context/kafka-rpc-proxy';
 import { RpcContextCreator } from './context/rpc-context-creator';
 import { RpcProxy } from './context/rpc-proxy';
+import { Transport } from './enums';
 import { CustomTransportStrategy } from './interfaces';
 import { ListenersController } from './listeners-controller';
 import { Server } from './server/server';
@@ -23,8 +25,13 @@ export class MicroservicesModule {
   private readonly clientsContainer = new ClientsContainer();
   private listenersController: ListenersController;
 
-  public register(container: NestContainer, config: ApplicationConfig) {
-    const rpcProxy = new RpcProxy();
+  public register(
+    container: NestContainer,
+    config: ApplicationConfig,
+    transport?: Transport,
+  ) {
+    const rpcProxy =
+      transport === Transport.KAFKA ? new KafkaRpcProxy() : new RpcProxy();
     const exceptionFiltersContext = new ExceptionFiltersContext(
       container,
       config,

packages/microservices/nest-microservice.ts

@@ -46,7 +46,12 @@ export class NestMicroservice
   ) {
     super(container);
 
-    this.microservicesModule.register(container, this.applicationConfig);
+    this.microservicesModule.register(
+      container,
+      this.applicationConfig,
+      // @ts-expect-error transport does not exist on type of config??
+      config.transport,
+    );
     this.createServer(config);
     this.selectContextModule();
   }