feat(microservices): add commit offset to kafka context

Provide access to kafkajs commitOffset by using @Ctx() decorator when handeling an event.

This commit closes Kafka commitOffsets #9283.

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/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;
+    });
+  });
 });