update sequential

Author: andrechristikan

src/configs/kafka.config.ts

@@ -23,6 +23,8 @@ export default registerAs(
             maxBytes: bytes('10mb'), // 5mb
             maxWaitTimeInMs: ms('5s'), // 5s
 
+            maxInFlightRequests: null, // set this to make customer guaranteed sequential
+
             retry: {
                 maxRetryTime: ms('60s'), // 30s
                 initialRetryTime: ms('0.3s'), // 3s

src/kafka.ts

@@ -3,7 +3,7 @@ import { ConfigService } from '@nestjs/config';
 import { NestApplication } from '@nestjs/core';
 import { MicroserviceOptions, Transport } from '@nestjs/microservices';
 import { ConsumerConfig, ConsumerSubscribeTopics } from 'kafkajs';
-import { KAFKA_TOPICS } from 'src/kafka/constants/kafka.constant';
+import { ENUM_KAFKA_TOPICS } from 'src/kafka/constants/kafka.topic.constant';
 
 export default async function (app: NestApplication) {
     const configService = app.get(ConfigService);
@@ -22,7 +22,7 @@ export default async function (app: NestApplication) {
         'kafka.allowAutoTopicCreation'
     );
     const subscribe: ConsumerSubscribeTopics = {
-        topics: KAFKA_TOPICS,
+        topics: Object.values(ENUM_KAFKA_TOPICS),
         ...configService.get<ConsumerSubscribeTopics>(
             'kafka.consumerSubscribe'
         ),

src/kafka/constants/kafka.constant.ts

@@ -1,11 +1,11 @@
+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_TOPICS: string[] = [
-    ...new Set(Object.values(ENUM_KAFKA_TOPICS)),
-];
-
-export const KAFKA_TOPICS_REPLY: string[] = [
-    ...new Set(Object.values(ENUM_KAFKA_TOPICS).map((val) => `${val}.reply`)),
-];
+export const KafkaCreateTopis: IKafkaCreateTopic[] = Object.values(
+    ENUM_KAFKA_TOPICS
+).map((val) => ({
+    topic: val,
+    topicReply: `${val}.reply`,
+}));

src/kafka/constants/kafka.topic.constant.ts

@@ -1,4 +1,4 @@
 export enum ENUM_KAFKA_TOPICS {
-    ACK_SUCCESS = 'nestjs.ack.success2',
-    ACK_ERROR = 'nestjs.ack.error2',
+    ACK_SUCCESS = 'nestjs.ack.success',
+    ACK_ERROR = 'nestjs.ack.error',
 }

src/kafka/interfaces/kafka.interface.ts

@@ -26,3 +26,10 @@ export interface IKafkaErrorException {
     errors: ValidationError;
     statusHttp: HttpStatus;
 }
+
+export interface IKafkaCreateTopic {
+    topic: string;
+    topicReply: string;
+    partition?: number;
+    replicationFactor?: number;
+}

src/kafka/interfaces/kafka.producer-service.interface.ts

@@ -17,5 +17,17 @@ export interface IKafkaProducerService {
         options?: IKafkaProducerMessageOptions
     ): void;
 
+    sendSequential<T, N>(
+        topic: string,
+        data: T,
+        options?: IKafkaProducerSendMessageOptions
+    ): Promise<IKafkaMessage<N> | N>;
+
+    emitSequential<T>(
+        topic: string,
+        data: T,
+        options?: IKafkaProducerMessageOptions
+    ): void;
+
     createId(): string;
 }

src/kafka/kafka.module.ts

@@ -1,11 +1,8 @@
 import { DynamicModule, Global, Module } from '@nestjs/common';
 import { ConfigModule, ConfigService } from '@nestjs/config';
 import { ClientsModule, Transport } from '@nestjs/microservices';
-import { ConsumerSubscribeTopics, ProducerConfig } from 'kafkajs';
-import {
-    KAFKA_PRODUCER_SERVICE_NAME,
-    KAFKA_TOPICS_REPLY,
-} from './constants/kafka.constant';
+import { ProducerConfig } from 'kafkajs';
+import { KAFKA_PRODUCER_SERVICE_NAME } from './constants/kafka.constant';
 import { KafkaRouterModule } from './router/kafka.router.module';
 import { KafkaAdminService } from './services/kafka.admin.service';
 import { KafkaProducerService } from './services/kafka.producer.service';

src/kafka/services/kafka.admin.service.ts

@@ -2,17 +2,17 @@ import { Injectable, OnModuleDestroy, OnModuleInit } from '@nestjs/common';
 import { Admin, ITopicConfig, Kafka, KafkaConfig } from 'kafkajs';
 import { Logger } from '@nestjs/common/services/logger.service';
 import { ConfigService } from '@nestjs/config';
-import { KAFKA_TOPICS, KAFKA_TOPICS_REPLY } from '../constants/kafka.constant';
 import { IKafkaAdminService } from 'src/kafka/interfaces/kafka.admin-service.interface';
+import { IKafkaCreateTopic } from 'src/kafka/interfaces/kafka.interface';
+import { KafkaCreateTopis } from 'src/kafka/constants/kafka.constant';
 
 @Injectable()
 export class KafkaAdminService
     implements IKafkaAdminService, OnModuleInit, OnModuleDestroy
 {
     private readonly kafka: Kafka;
     private readonly admin: Admin;
-    private readonly topics: string[];
-    private readonly topicsReply: string[];
+    private readonly topics: IKafkaCreateTopic[];
     private readonly brokers: string[];
     private readonly clientId: string;
     private readonly kafkaOptions: KafkaConfig;
@@ -24,8 +24,7 @@ export class KafkaAdminService
         this.clientId = this.configService.get<string>('kafka.admin.clientId');
         this.brokers = this.configService.get<string[]>('kafka.brokers');
 
-        this.topics = KAFKA_TOPICS;
-        this.topicsReply = KAFKA_TOPICS_REPLY;
+        this.topics = KafkaCreateTopis;
 
         this.kafkaOptions = {
             clientId: this.clientId,
@@ -74,27 +73,30 @@ export class KafkaAdminService
 
     async createTopics(): Promise<boolean> {
         this.logger.log(`Topics ${this.topics}`);
-        this.logger.log(`Topics Reply ${this.topicsReply}`);
 
         const currentTopic: string[] = await this.getAllTopicUnique();
         const data: ITopicConfig[] = [];
 
         for (const topic of this.topics) {
-            if (!currentTopic.includes(topic)) {
+            const partition: number = topic.partition ?? this.defaultPartition;
+            const replicationFactor: number =
+                topic.replicationFactor &&
+                topic.replicationFactor <= this.brokers.length
+                    ? topic.replicationFactor
+                    : this.brokers.length;
+            if (!currentTopic.includes(topic.topic)) {
                 data.push({
-                    topic,
-                    numPartitions: this.defaultPartition,
-                    replicationFactor: this.brokers.length,
+                    topic: topic.topic,
+                    numPartitions: partition,
+                    replicationFactor: replicationFactor,
                 });
             }
-        }
 
-        for (const replyTopic of this.topicsReply) {
-            if (!currentTopic.includes(replyTopic)) {
+            if (!currentTopic.includes(topic.topicReply)) {
                 data.push({
-                    topic: replyTopic,
-                    numPartitions: this.defaultPartition,
-                    replicationFactor: this.brokers.length,
+                    topic: topic.topicReply,
+                    numPartitions: partition,
+                    replicationFactor,
                 });
             }
         }
@@ -117,14 +119,12 @@ export class KafkaAdminService
         const data = [];
 
         for (const topic of this.topics) {
-            if (currentTopic.includes(topic)) {
-                data.push(topic);
+            if (currentTopic.includes(topic.topic)) {
+                data.push(topic.topic);
             }
-        }
 
-        for (const replyTopic of this.topicsReply) {
-            if (currentTopic.includes(replyTopic)) {
-                data.push(replyTopic);
+            if (currentTopic.includes(topic.topicReply)) {
+                data.push(topic.topicReply);
             }
         }
 

src/kafka/services/kafka.producer.service.ts

@@ -15,10 +15,8 @@ import {
     IKafkaProducerSendMessageOptions,
 } from 'src/kafka/interfaces/kafka.interface';
 import { IKafkaProducerService } from 'src/kafka/interfaces/kafka.producer-service.interface';
-import {
-    KAFKA_PRODUCER_SERVICE_NAME,
-    KAFKA_TOPICS,
-} from '../constants/kafka.constant';
+import { KAFKA_PRODUCER_SERVICE_NAME } from '../constants/kafka.constant';
+import { ENUM_KAFKA_TOPICS } from 'src/kafka/constants/kafka.topic.constant';
 
 @Injectable()
 export class KafkaProducerService
@@ -40,7 +38,7 @@ export class KafkaProducerService
     }
 
     async onApplicationBootstrap(): Promise<void> {
-        KAFKA_TOPICS.forEach((topic) =>
+        Object.values(ENUM_KAFKA_TOPICS).forEach((topic) =>
             this.clientKafka.subscribeToResponseOf(topic)
         );
 
@@ -91,6 +89,48 @@ export class KafkaProducerService
         return;
     }
 
+    async sendSequential<T, N>(
+        topic: string,
+        data: T,
+        options?: IKafkaProducerSendMessageOptions
+    ): Promise<IKafkaMessage<N> | N> {
+        const message: IKafkaMessage<T> = {
+            key: `${topic}-sequential-key`,
+            value: data,
+            headers: options && options.headers ? options.headers : undefined,
+        };
+
+        const send = await firstValueFrom(
+            this.clientKafka
+                .send<any, string>(topic, JSON.stringify(message))
+                .pipe(timeout(this.timeout))
+        );
+
+        if (send.error) {
+            throw send.error;
+        }
+
+        return options && options.raw ? send : send.value;
+    }
+
+    emitSequential<T>(
+        topic: string,
+        data: T,
+        options?: IKafkaProducerMessageOptions
+    ): void {
+        const message: IKafkaMessage<T> = {
+            key: `${topic}-sequential-key`,
+            value: data,
+            headers: options && options.headers ? options.headers : undefined,
+        };
+
+        this.clientKafka
+            .emit<any, string>(topic, JSON.stringify(message))
+            .pipe(timeout(this.timeout));
+
+        return;
+    }
+
     createId(): string {
         const rand: string = this.helperStringService.random(10);
         const timestamp = `${this.helperDateService.timestamp()}`;