Fix kafka retries when error, and dont commit message. That behaviour will break some workflow, so we add commit offset intercetor in first runing

Author: andrechristikan

.env.example

@@ -54,6 +54,5 @@ AWS_S3_BUCKET=baibay-development
 KAFKA_CLIENT_ID=KAFKA_ACK
 KAFKA_ADMIN_CLIENT_ID=KAFKA_ADMIN_ACK
 KAFKA_BROKERS=localhost:9092
-KAFKA_PRODUCER_ENABLE=true
 KAFKA_CONSUMER_ENABLE=true
 KAFKA_CONSUMER_GROUP=nestjs.ack

README.md

@@ -8,7 +8,28 @@ Fork from [ack-nestjs-boilerplate](https://github.com/andrechristikan/ack-nestjs
 
 ## Instructions
 
-Before run, you must to create the topics with `kafka-topics cli`. Or you can run `yarn kafka:create-topics`.
+Before run, you must to create the topics with `kafka-topics cli`. Or you can run `yarn kafka:create-topics`. (Point 9 at Behaviour)
+
+
+## Behaviour
+
+1. Producer and consumer will use `nestjs/microservice`
+2. Producer will always imported as global module
+3. Message not commit offset in error
+4. Optional commit offset in first running
+5. Default kafka will create topic with 3 partition and 3 replication factor
+6. For guarantee sequential 
+    - Create topic with `1 partition` and `1 replication factor`
+    - Set `maxInFlightRequests` to `1` in consumer config
+    - Use `producerSendSequential` or `produceEmitSequential`
+7. `emit` means that whether or not you explicitly subscribe to the observable, the proxy will immediately try to deliver the event.
+8. `send` means that you have to explicitly subscribe to it before the message will be sent.
+9. `acks` will depends on kafka setting. Unfortunately, nestjs can not set acks when produce some message.
+10. Topics in `./src/kafka/constants/kafka.topic.constant.ts` must created before running.
+11. All kafka request will validate with `class-validation` if you set the dto class.
+12. Put KafkaController into `./src/kafka/router/kafka.router.module.ts`, so you can use `env.KAFKA_CONSUMER_ENABLE` for on/off
+
+
 
 ## Next
 

src/common/common.module.ts

@@ -158,7 +158,6 @@ import { APP_LANGUAGE } from 'src/app/constants/app.constant';
                 KAFKA_BROKERS: Joi.string()
                     .default('localhost:9092')
                     .required(),
-                    KAFKA_PRODUCER_ENABLE: Joi.boolean().default(true).required(),
                 KAFKA_CONSUMER_ENABLE: Joi.boolean().default(true).required(),
                 KAFKA_CONSUMER_GROUP: Joi.string()
                     .default('nestjs.ack')

src/configs/kafka.config.ts

@@ -36,7 +36,6 @@ export default registerAs(
         },
 
         // producer
-        producerEnable: process.env.KAFKA_PRODUCER_ENABLE === 'true',
         producer: {
             createPartitioner: Partitioners.LegacyPartitioner,
             transactionTimeout: ms('100s'), // 30000 .. 60000

src/kafka/constants/kafka.constant.ts

@@ -1,11 +1,13 @@
 import { IKafkaCreateTopic } from 'src/kafka/interfaces/kafka.interface';
 import { ENUM_KAFKA_TOPICS } from './kafka.topic.constant';
 
-export const KAFKA_PRODUCER_SERVICE_NAME = 'KAFKA_PRODUCER_SERVICE';
+export const KAFKA_SERVICE_NAME = 'KAFKA_SERVICE';
 
 export const KafkaCreateTopis: IKafkaCreateTopic[] = Object.values(
     ENUM_KAFKA_TOPICS
 ).map((val) => ({
     topic: val,
     topicReply: `${val}.reply`,
+    // partition?: number;
+    // replicationFactor?: number;
 }));

src/kafka/controllers/kafka.controller.ts

@@ -4,6 +4,7 @@ import { Logger } from 'src/common/logger/decorators/logger.decorator';
 import { IResponse } from 'src/common/response/interfaces/response.interface';
 import { ENUM_KAFKA_TOPICS } from 'src/kafka/constants/kafka.topic.constant';
 import {
+    MessageCommitOffsetInFirstRunning,
     MessageTopic,
     MessageValue,
 } from 'src/kafka/decorators/kafka.decorator';
@@ -12,6 +13,7 @@ import { KafkaDto } from 'src/kafka/dtos/kafka.dto';
 @Controller()
 export class KafkaController {
     @Logger(ENUM_LOGGER_ACTION.TEST, { tags: ['helloKafka'] })
+    @MessageCommitOffsetInFirstRunning()
     @MessageTopic(ENUM_KAFKA_TOPICS.ACK_SUCCESS)
     async helloKafka(
         @MessageValue() value: Record<string, any>
@@ -20,6 +22,7 @@ export class KafkaController {
     }
 
     @Logger(ENUM_LOGGER_ACTION.TEST, { tags: ['helloKafkaError'] })
+    @MessageCommitOffsetInFirstRunning()
     @MessageTopic(ENUM_KAFKA_TOPICS.ACK_ERROR)
     async errorKafka(@MessageValue() value: KafkaDto): Promise<IResponse> {
         return value;

src/kafka/controllers/kafka.test.controller.ts

@@ -1,29 +1,27 @@
-import { Controller, Get, Optional, VERSION_NEUTRAL } from '@nestjs/common';
+import { Controller, Get, VERSION_NEUTRAL } from '@nestjs/common';
 import { ApiExcludeController } from '@nestjs/swagger';
 import { Types } from 'mongoose';
 import { ENUM_LOGGER_ACTION } from 'src/common/logger/constants/logger.enum.constant';
 import { Logger } from 'src/common/logger/decorators/logger.decorator';
 import { Response } from 'src/common/response/decorators/response.decorator';
 import { IResponse } from 'src/common/response/interfaces/response.interface';
 import { ENUM_KAFKA_TOPICS } from 'src/kafka/constants/kafka.topic.constant';
-import { KafkaException } from 'src/kafka/error/exceptions/kafka.exception';
-import { KafkaProducerService } from 'src/kafka/services/kafka.producer.service';
+import { KafkaHttpException } from 'src/kafka/error/exceptions/kafka.http-exception';
+import { KafkaService } from 'src/kafka/services/kafka.service';
 
 @ApiExcludeController()
 @Controller({
     version: VERSION_NEUTRAL,
     path: '/kafka',
 })
 export class KafkaTestController {
-    constructor(
-        @Optional() private readonly kafkaProducerService: KafkaProducerService
-    ) {}
+    constructor(private readonly kafkaService: KafkaService) {}
 
     @Response('test.helloKafka')
     @Logger(ENUM_LOGGER_ACTION.TEST, { tags: ['testKafka'] })
     @Get('/')
     async helloKafka(): Promise<IResponse> {
-        const response = await this.kafkaProducerService.send(
+        const response = await this.kafkaService.produceSend(
             ENUM_KAFKA_TOPICS.ACK_SUCCESS,
             {
                 test: 'test',
@@ -53,7 +51,7 @@ export class KafkaTestController {
     @Get('/error')
     async helloKafkaError(): Promise<IResponse> {
         try {
-            const response = await this.kafkaProducerService.send(
+            const response = await this.kafkaService.produceSend(
                 ENUM_KAFKA_TOPICS.ACK_ERROR,
                 {
                     testNumber: [],
@@ -78,7 +76,7 @@ export class KafkaTestController {
 
             return response;
         } catch (err: any) {
-            throw new KafkaException(err);
+            throw new KafkaHttpException(err);
         }
     }
 }

src/kafka/decorators/kafka.decorator.ts

@@ -16,6 +16,7 @@ import { KafkaErrorFilter } from '../error/filters/kafka.error.filter';
 import { KafkaResponseInterceptor } from '../interceptors/kafka.response.interceptor';
 import { KafkaResponseTimeoutInterceptor } from '../interceptors/kafka.response.timeout.interceptor';
 import { KafkaValidationPipe } from '../pipes/kafka.validation.pipe';
+import { KafkaCommitOffsetFirstInterceptor } from 'src/kafka/interceptors/kafka.commit-offset-first.interceptor';
 
 export function MessageTopic(topic: string): any {
     return applyDecorators(
@@ -46,3 +47,7 @@ export const MessageKey = createParamDecorator<string>(
         return key;
     }
 );
+
+export function MessageCommitOffsetInFirstRunning(): any {
+    return applyDecorators(UseInterceptors(KafkaCommitOffsetFirstInterceptor));
+}

src/kafka/error/exceptions/kafka.exception.ts renamed to src/kafka/error/exceptions/kafka.http-exception.ts

@@ -2,7 +2,7 @@ import { HttpException, HttpStatus } from '@nestjs/common';
 import { ENUM_ERROR_STATUS_CODE_ERROR } from 'src/common/error/constants/error.status-code.constant';
 import { IKafkaErrorException } from 'src/kafka/interfaces/kafka.interface';
 
-export class KafkaException extends HttpException {
+export class KafkaHttpException extends HttpException {
     constructor(exception: IKafkaErrorException) {
         if (
             'message' in exception &&

src/kafka/interceptors/kafka.commit-offset-first.interceptor.ts

@@ -0,0 +1,32 @@
+import {
+    Injectable,
+    NestInterceptor,
+    ExecutionContext,
+    CallHandler,
+} from '@nestjs/common';
+import { RpcArgumentsHost } from '@nestjs/common/interfaces';
+import { KafkaContext } from '@nestjs/microservices';
+import { Observable } from 'rxjs';
+import { map } from 'rxjs/operators';
+import { KafkaService } from 'src/kafka/services/kafka.service';
+
+@Injectable()
+export class KafkaCommitOffsetFirstInterceptor
+    implements NestInterceptor<Promise<any>>
+{
+    constructor(private readonly kafkaService: KafkaService) {}
+
+    async intercept(
+        context: ExecutionContext,
+        next: CallHandler
+    ): Promise<Observable<Promise<any> | string>> {
+        const ctx: RpcArgumentsHost = context.switchToRpc();
+        const kafkaContext = ctx.getContext<KafkaContext>();
+
+        try {
+            await this.kafkaService.commitOffsets(kafkaContext);
+        } catch (error: unknown) {}
+
+        return next.handle();
+    }
+}