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

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

@@ -0,0 +1,14 @@
+import { Observable } from 'rxjs';
+import { RpcExceptionsHandler } from '../exceptions/rpc-exceptions-handler';
+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/ctx-host/kafka.context.ts

@@ -1,7 +1,12 @@
 import { KafkaMessage } from '../external/kafka.interface';
 import { BaseRpcContext } from './base-rpc.context';
 
-type KafkaContextArgs = [KafkaMessage, number, string];
+type KafkaContextArgs = [
+  message: KafkaMessage,
+  partition: number,
+  topic: string,
+  commitOffset: () => Promise<void>,
+];
 
 export class KafkaContext extends BaseRpcContext<KafkaContextArgs> {
   constructor(args: KafkaContextArgs) {
@@ -28,4 +33,11 @@ export class KafkaContext extends BaseRpcContext<KafkaContextArgs> {
   getTopic() {
     return this.args[2];
   }
+
+  /**
+   * Commit the offset of this message.
+   */
+  commitOffset() {
+    return this.args[3]();
+  }
 }

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();
   }

packages/microservices/server/server-kafka.ts

@@ -1,6 +1,5 @@
 import { Logger } from '@nestjs/common/services/logger.service';
 import { isNil } from '@nestjs/common/utils/shared.utils';
-import { Observable } from 'rxjs';
 import {
   KAFKA_DEFAULT_BROKER,
   KAFKA_DEFAULT_CLIENT,
@@ -162,6 +161,14 @@ export class ServerKafka extends Server implements CustomTransportStrategy {
       rawMessage,
       payload.partition,
       payload.topic,
+      () =>
+        this.consumer.commitOffsets([
+          {
+            offset: (parseInt(payload.message.offset, 10) + 1).toString(),
+            partition: payload.partition,
+            topic: payload.topic,
+          },
+        ]),
     ]);
     // if the correlation id or reply topic is not set
     // then this is an event (events could still have correlation id)

packages/microservices/test/ctx-host/kafka.context.spec.ts

@@ -1,13 +1,17 @@
 import { expect } from 'chai';
+import * as sinon from 'sinon';
 import { KafkaContext } from '../../ctx-host';
 import { KafkaMessage } from '../../external/kafka.interface';
 
 describe('KafkaContext', () => {
-  const args = ['test', { test: true }];
+  const testFunc = sinon.spy();
+  const args = ['test', { test: true }, undefined, testFunc];
   let context: KafkaContext;
 
   beforeEach(() => {
-    context = new KafkaContext(args as [KafkaMessage, number, string]);
+    context = new KafkaContext(
+      args as [KafkaMessage, number, string, () => Promise<void>],
+    );
   });
   describe('getTopic', () => {
     it('should return topic', () => {
@@ -24,4 +28,10 @@ describe('KafkaContext', () => {
       expect(context.getMessage()).to.be.eql(args[0]);
     });
   });
+  describe('commitOffset', () => {
+    it('should be called once', () => {
+      context.commitOffset();
+      expect(testFunc.called).to.be.true;
+    });
+  });
 });