Kafka

Kafka là một nền tảng streaming phân tán, mã nguồn mở, có ba khả năng chính:

• Publish và subscribe các stream record, tương tự như message queue hoặc hệ thống enterprise messaging.
• Lưu trữ các stream record theo cách bền vững, chịu lỗi (fault-tolerant).
• Xử lý các stream record ngay khi chúng xảy ra.

Dự án Kafka nhằm cung cấp một nền tảng thống nhất, throughput cao, độ trễ thấp để xử lý luồng dữ liệu thời gian thực. Nó tích hợp rất tốt với Apache Storm và Spark để phân tích dữ liệu streaming thời gian thực.

Cài đặt

Để bắt đầu xây dựng microservice dựa trên Kafka, trước tiên hãy cài package cần thiết:

$ npm i --save kafkajs

Tổng quan

Giống như các cách triển khai transport layer microservice khác của Nest, bạn chọn cơ chế transporter Kafka bằng thuộc tính transport của đối tượng options truyền vào phương thức createMicroservice(), cùng với một thuộc tính options tùy chọn, như dưới đây:

const app = await NestFactory.createMicroservice<MicroserviceOptions>(AppModule, {
  transport: Transport.KAFKA,
  options: {
    client: {
      brokers: ['localhost:9092'],
    }
  }
});

const app = await NestFactory.createMicroservice(AppModule, {
  transport: Transport.KAFKA,
  options: {
    client: {
      brokers: ['localhost:9092'],
    }
  }
});

Options

Thuộc tính options đặc thù theo transporter được chọn. Transporter Kafka expose các thuộc tính được mô tả dưới đây.

Client

Có một khác biệt nhỏ ở Kafka so với các transporter microservice khác. Thay vì dùng class ClientProxy, ta dùng class ClientKafkaProxy.

Giống các transporter microservice khác, bạn có nhiều cách để tạo instance ClientKafkaProxy.

Một cách để tạo instance là dùng ClientsModule. Để tạo client instance với ClientsModule, import nó và dùng phương thức register() để truyền một đối tượng options với các thuộc tính giống như trong phương thức createMicroservice() ở trên, cùng với thuộc tính name dùng làm injection token. Đọc thêm về ClientsModule tại đây.

@Module({
  imports: [
    ClientsModule.register([
      {
        name: 'HERO_SERVICE',
        transport: Transport.KAFKA,
        options: {
          client: {
            clientId: 'hero',
            brokers: ['localhost:9092'],
          },
          consumer: {
            groupId: 'hero-consumer'
          }
        }
      },
    ]),
  ]
  ...
})

Các cách khác để tạo client (ClientProxyFactory hoặc @Client()) cũng có thể dùng được. Bạn có thể đọc thêm về chúng tại đây.

Dùng decorator @Client() như sau:

@Client({
  transport: Transport.KAFKA,
  options: {
    client: {
      clientId: 'hero',
      brokers: ['localhost:9092'],
    },
    consumer: {
      groupId: 'hero-consumer'
    }
  }
})
client: ClientKafkaProxy;

Message pattern

Message pattern của Kafka microservice sử dụng hai topic cho kênh request và kênh reply. Phương thức ClientKafkaProxy.send() gửi message kèm return address bằng cách gắn một correlation id, reply topic, và reply partition vào request message. Điều này yêu cầu instance ClientKafkaProxy phải subscribe vào reply topic và được gán vào ít nhất một partition trước khi gửi message.

Do đó, bạn cần có ít nhất một reply topic partition cho mỗi Nest application đang chạy. Ví dụ, nếu bạn đang chạy 4 Nest application nhưng reply topic chỉ có 3 partition, thì 1 trong 4 Nest application đó sẽ bị lỗi khi cố gửi message.

Khi các instance ClientKafkaProxy mới được khởi chạy, chúng sẽ join vào consumer group và subscribe vào các topic tương ứng. Quá trình này kích hoạt một lần rebalance các partition của topic được gán cho các consumer trong consumer group.

Thông thường, partition của topic được gán bằng round robin partitioner, gán partition cho tập hợp consumer được sắp xếp theo tên consumer (tên này được đặt ngẫu nhiên khi application khởi chạy). Tuy nhiên, khi một consumer mới join vào consumer group, consumer mới đó có thể được đặt ở bất kỳ vị trí nào trong tập hợp consumer. Điều này tạo ra tình trạng các consumer đã tồn tại từ trước có thể bị gán partition khác nếu consumer đó nằm sau consumer mới trong thứ tự sắp xếp. Kết quả là các consumer bị gán partition khác sẽ mất các response message của các request đã gửi trước khi rebalance xảy ra.

Để ngăn các consumer ClientKafkaProxy mất response message, Nest sử dụng một custom partitioner tích hợp sẵn, đặc thù cho Nest. Custom partitioner này gán partition cho tập hợp consumer được sắp xếp theo timestamp độ phân giải cao (process.hrtime()) được đặt khi application khởi chạy.

Pattern dạng biểu thức chính quy (regular expression)

KafkaJS hỗ trợ subscribe topic bằng biểu thức chính quy (regular expression), và từ NestJS v12 bạn có thể truyền trực tiếp một RegExp vào @MessagePattern() hoặc @EventPattern(). Nest giữ nguyên pattern, chuyển tiếp nó tới lệnh gọi subscribe() của KafkaJS, và dùng cơ chế khớp biểu thức chính quy khi phân giải handler cho một topic gửi đến.

@EventPattern(/^hero\..+$/)
handleHeroEvents(@Payload() data: any, @Ctx() context: KafkaContext) {
  console.log(`Topic: ${context.getTopic()}`);
}

@Bind(Payload(), Ctx())
@EventPattern(/^hero\..+$/)
handleHeroEvents(data, context) {
  console.log(`Topic: ${context.getTopic()}`);
}

Điều này subscribe handler vào mọi topic khớp với biểu thức trên - hero.kill.dragon, hero.rescue.villager, v.v. - mà không cần đăng ký từng topic một cách rõ ràng. Dùng context.getTopic() để biết topic nào thực sự đã gửi message.

Đăng ký nhận response message

Class ClientKafkaProxy cung cấp phương thức subscribeToResponseOf(). Phương thức subscribeToResponseOf() nhận tên topic của một request làm tham số và thêm tên reply topic tương ứng vào một tập hợp reply topic. Phương thức này là bắt buộc khi triển khai message pattern.

onModuleInit() {
  this.client.subscribeToResponseOf('hero.kill.dragon');
}

Nếu instance ClientKafkaProxy được tạo bất đồng bộ (asynchronously), phương thức subscribeToResponseOf() phải được gọi trước khi gọi phương thức connect().

async onModuleInit() {
  this.client.subscribeToResponseOf('hero.kill.dragon');
  await this.client.connect();
}

Incoming (message đến)

Nest nhận các Kafka message đến dưới dạng một object có các thuộc tính key, value, và headers với giá trị kiểu Buffer. Sau đó Nest parse các giá trị này bằng cách chuyển buffer thành string. Nếu string đó “giống object” (object-like), Nest sẽ thử parse string đó dưới dạng JSON. Giá trị value sau đó được truyền tới handler tương ứng.

Outgoing (message gửi đi)

Nest gửi các Kafka message đi sau một quá trình tuần tự hóa (serialization) khi publish event hoặc gửi message. Quá trình này diễn ra trên các tham số truyền vào phương thức emit() và send() của ClientKafkaProxy, hoặc trên giá trị trả về từ phương thức được gắn @MessagePattern. Quá trình serialization này “chuyển thành string” các object không phải string hoặc buffer bằng cách dùng JSON.stringify() hoặc phương thức toString() trên prototype.

@Controller()
export class HeroesController {
  @MessagePattern('hero.kill.dragon')
  killDragon(@Payload() message: KillDragonMessage): any {
    const dragonId = message.dragonId;
    const items = [
      { id: 1, name: 'Mythical Sword' },
      { id: 2, name: 'Key to Dungeon' },
    ];
    return items;
  }
}

Message gửi đi cũng có thể được gắn key bằng cách truyền một object có thuộc tính key và value. Việc gắn key cho message quan trọng để đáp ứng yêu cầu co-partitioning.

@Controller()
export class HeroesController {
  @MessagePattern('hero.kill.dragon')
  killDragon(@Payload() message: KillDragonMessage): any {
    const realm = 'Nest';
    const heroId = message.heroId;
    const dragonId = message.dragonId;

    const items = [
      { id: 1, name: 'Mythical Sword' },
      { id: 2, name: 'Key to Dungeon' },
    ];

    return {
      headers: {
        realm
      },
      key: heroId,
      value: items
    }
  }
}

Ngoài ra, message truyền theo định dạng này còn có thể chứa header tùy chỉnh được đặt trong thuộc tính hash headers. Giá trị của thuộc tính hash header phải thuộc kiểu string hoặc kiểu Buffer.

@Controller()
export class HeroesController {
  @MessagePattern('hero.kill.dragon')
  killDragon(@Payload() message: KillDragonMessage): any {
    const realm = 'Nest';
    const heroId = message.heroId;
    const dragonId = message.dragonId;

    const items = [
      { id: 1, name: 'Mythical Sword' },
      { id: 2, name: 'Key to Dungeon' },
    ];

    return {
      headers: {
        kafka_nestRealm: realm
      },
      key: heroId,
      value: items
    }
  }
}

Dựa trên event (event-based)

Trong khi phương thức request-response phù hợp để trao đổi message giữa các service, nó lại kém phù hợp hơn khi kiểu message của bạn là event-based (kiểu này ngược lại lại rất phù hợp với Kafka) - khi bạn chỉ muốn publish event mà không cần chờ response. Trong trường hợp đó, bạn không muốn chịu overhead mà request-response yêu cầu để duy trì hai topic.

Xem hai phần sau để tìm hiểu thêm về điều này: Tổng quan: Dựa trên event và Tổng quan: Publish event.

Context

Trong các tình huống phức tạp hơn, bạn có thể cần truy cập thêm thông tin về request đến. Khi dùng transporter Kafka, bạn có thể truy cập object KafkaContext.

@MessagePattern('hero.kill.dragon')
killDragon(@Payload() message: KillDragonMessage, @Ctx() context: KafkaContext) {
  console.log(`Topic: ${context.getTopic()}`);
}

@Bind(Payload(), Ctx())
@MessagePattern('hero.kill.dragon')
killDragon(message, context) {
  console.log(`Topic: ${context.getTopic()}`);
}

Để truy cập object IncomingMessage gốc của Kafka, dùng phương thức getMessage() của object KafkaContext, như sau:

@MessagePattern('hero.kill.dragon')
killDragon(@Payload() message: KillDragonMessage, @Ctx() context: KafkaContext) {
  const originalMessage = context.getMessage();
  const partition = context.getPartition();
  const { headers, timestamp } = originalMessage;
}

@Bind(Payload(), Ctx())
@MessagePattern('hero.kill.dragon')
killDragon(message, context) {
  const originalMessage = context.getMessage();
  const partition = context.getPartition();
  const { headers, timestamp } = originalMessage;
}

Trong đó IncomingMessage tuân theo interface sau:

interface IncomingMessage {
  topic: string;
  partition: number;
  timestamp: string;
  size: number;
  attributes: number;
  offset: string;
  key: any;
  value: any;
  headers: Record<string, any>;
}

Nếu handler của bạn có thời gian xử lý chậm cho mỗi message nhận được, bạn nên xem xét dùng callback heartbeat. Để lấy hàm heartbeat, dùng phương thức getHeartbeat() của KafkaContext, như sau:

@MessagePattern('hero.kill.dragon')
async killDragon(@Payload() message: KillDragonMessage, @Ctx() context: KafkaContext) {
  const heartbeat = context.getHeartbeat();

  // Thực hiện một phần xử lý chậm
  await doWorkPart1();

  // Gửi heartbeat để không vượt quá sessionTimeout
  await heartbeat();

  // Thực hiện phần xử lý chậm tiếp theo
  await doWorkPart2();
}

Quy ước đặt tên

Các thành phần Kafka microservice thêm mô tả vai trò tương ứng của chúng vào cuối các tùy chọn client.clientId và consumer.groupId để tránh xung đột giữa thành phần client và server của Nest microservice. Theo mặc định, các thành phần ClientKafkaProxy thêm -client và các thành phần ServerKafka thêm -server vào cả hai tùy chọn này. Hãy chú ý cách các giá trị được cung cấp dưới đây bị biến đổi theo cách đó (như trong comment).

const app = await NestFactory.createMicroservice<MicroserviceOptions>(AppModule, {
  transport: Transport.KAFKA,
  options: {
    client: {
      clientId: 'hero', // hero-server
      brokers: ['localhost:9092'],
    },
    consumer: {
      groupId: 'hero-consumer' // hero-consumer-server
    },
  }
});

Và đối với client:

@Client({
  transport: Transport.KAFKA,
  options: {
    client: {
      clientId: 'hero', // hero-client
      brokers: ['localhost:9092'],
    },
    consumer: {
      groupId: 'hero-consumer' // hero-consumer-client
    }
  }
})
client: ClientKafkaProxy;

Vì message pattern của Kafka microservice sử dụng hai topic cho kênh request và kênh reply, một reply pattern cần được suy ra từ request topic. Theo mặc định, tên của reply topic là ghép giữa tên request topic với hậu tố .reply.

onModuleInit() {
  this.client.subscribeToResponseOf('hero.get'); // hero.get.reply
}

Exception có thể retry

Tương tự các transporter khác, tất cả exception không được xử lý (unhandled) đều tự động được bọc vào một RpcException và chuyển đổi sang định dạng “thân thiện với người dùng”. Tuy nhiên, có những trường hợp đặc biệt (edge-case) khi bạn muốn bỏ qua cơ chế này và để exception được driver kafkajs xử lý thay thế. Ném (throw) một exception khi đang xử lý một message sẽ chỉ định kafkajs retry (gửi lại) message đó, nghĩa là mặc dù handler của message (hoặc event) đã được kích hoạt, offset sẽ không được commit vào Kafka.

Để làm điều này, bạn có thể dùng một class riêng gọi là KafkaRetriableException, như sau:

throw new KafkaRetriableException('...');

Xử lý exception tùy chỉnh

Bên cạnh các cơ chế xử lý lỗi mặc định, bạn có thể tạo một Exception Filter tùy chỉnh cho các event Kafka để quản lý logic retry. Ví dụ dưới đây minh họa cách bỏ qua một event gặp lỗi sau một số lần retry có thể cấu hình:

import { Catch, ArgumentsHost, Logger } from '@nestjs/common';
import { BaseExceptionFilter } from '@nestjs/core';
import { KafkaContext } from '@nestjs/microservices';
import { Producer } from 'kafkajs';

@Catch()
export class KafkaMaxRetryExceptionFilter extends BaseExceptionFilter {
  private readonly logger = new Logger(KafkaMaxRetryExceptionFilter.name);

  constructor(
    private readonly producer: Producer,
    private readonly maxRetries: number,
    // Hàm tùy chọn được thực thi khi vượt quá số lần retry tối đa
    private readonly skipHandler?: (message: any) => Promise<void>,
  ) {
    super();
  }

  async catch(exception: unknown, host: ArgumentsHost) {
    const kafkaContext = host.switchToRpc().getContext<KafkaContext>();
    const message = kafkaContext.getMessage();
    const currentRetryCount = this.getRetryCountFromContext(kafkaContext);

    if (currentRetryCount >= this.maxRetries) {
      this.logger.warn(
        `Max retries (${
          this.maxRetries
        }) exceeded for message: ${JSON.stringify(message)}`,
      );

      if (this.skipHandler) {
        try {
          await this.skipHandler(message);
        } catch (err) {
          this.logger.error('Error in skipHandler:', err);
        }
      }

      try {
        await this.commitOffset(kafkaContext);
      } catch (commitError) {
        this.logger.error('Failed to commit offset:', commitError);
      }
      return; // Dừng việc lan truyền exception
    }

    // Republish message vào cùng topic với retry count đã tăng thêm
    try {
      await this.republishWithRetry(kafkaContext, currentRetryCount + 1);
      await this.commitOffset(kafkaContext);
    } catch (republishError) {
      this.logger.error(
        'Failed to republish message for retry:',
        republishError,
      );
      // Fall back về cơ chế xử lý exception mặc định
      super.catch(exception, host);
    }
  }

  private getRetryCountFromContext(context: KafkaContext): number {
    const headers = context.getMessage().headers || {};
    const retryHeader = headers['retry-count'];
    if (!retryHeader) {
      return 0;
    }
    // Giá trị header là Buffer, nên phải chuyển thành string trước
    const value = Buffer.isBuffer(retryHeader)
      ? retryHeader.toString()
      : String(retryHeader);
    return parseInt(value, 10) || 0;
  }

  private async republishWithRetry(
    context: KafkaContext,
    retryCount: number,
  ): Promise<void> {
    const topic = context.getTopic();
    const message = context.getMessage();

    await this.producer.send({
      topic,
      messages: [
        {
          key: message.key,
          value: message.value,
          headers: {
            ...message.headers,
            'retry-count': retryCount.toString(),
          },
        },
      ],
    });
  }

  private async commitOffset(context: KafkaContext): Promise<void> {
    const consumer = context.getConsumer();
    if (!consumer) {
      throw new Error('Consumer instance is not available from KafkaContext.');
    }

    const topic = context.getTopic();
    const partition = context.getPartition();
    const message = context.getMessage();
    const offset = message.offset;

    if (!topic || partition === undefined || offset === undefined) {
      throw new Error(
        'Incomplete Kafka message context for committing offset.',
      );
    }

    await consumer.commitOffsets([
      {
        topic,
        partition,
        // Khi commit offset, commit số tiếp theo (tức là offset hiện tại + 1)
        offset: (Number(offset) + 1).toString(),
      },
    ]);
  }
}

Filter này cung cấp một cách để retry xử lý một Kafka event tối đa một số lần có thể cấu hình. Khi xảy ra exception, nó republish message vào cùng topic với header retry-count đã tăng thêm, sau đó commit offset hiện tại. Khi đạt số lần retry tối đa, nó kích hoạt một skipHandler tùy chỉnh (nếu có) và commit offset, qua đó bỏ qua event gặp lỗi một cách hiệu quả. Điều này giúp các event tiếp theo được xử lý mà không bị gián đoạn.

Bạn có thể tích hợp filter này bằng cách đăng ký nó ở phạm vi global hoặc ở cấp controller. Lưu ý rằng bạn cần cung cấp một instance Kafka producer:

import { Inject, Injectable } from '@nestjs/common';
import { Producer } from 'kafkajs';

@Injectable()
export class AppKafkaRetryFilter extends KafkaMaxRetryExceptionFilter {
  constructor(@Inject('KAFKA_PRODUCER') producer: Producer) {
    super(producer, 5); // maxRetries = 5
  }
}

import { Inject, Injectable } from '@nestjs/common';

@Injectable()
export class AppKafkaRetryFilter extends KafkaMaxRetryExceptionFilter {
  constructor(@Inject('KAFKA_PRODUCER') producer) {
    super(producer, 5); // maxRetries = 5
  }
}

@Controller()
@UseFilters(AppKafkaRetryFilter)
export class MyEventHandler {
  @EventPattern('your-topic')
  async handleEvent(@Payload() data: any, @Ctx() context: KafkaContext) {
    // Logic xử lý event của bạn...
  }
}

@Controller()
@UseFilters(AppKafkaRetryFilter)
export class MyEventHandler {
  @Bind(Payload(), Ctx())
  @EventPattern('your-topic')
  async handleEvent(data, context) {
    // Logic xử lý event của bạn...
  }
}

Đảm bảo cung cấp Kafka producer trong module của bạn:

import { Kafka } from 'kafkajs';

@Module({
  providers: [
    AppKafkaRetryFilter,
    {
      provide: 'KAFKA_PRODUCER',
      useFactory: async () => {
        const kafka = new Kafka({ brokers: ['localhost:9092'] });
        const producer = kafka.producer();
        await producer.connect();
        return producer;
      },
    },
  ],
})
export class AppModule {}

Commit offset

Việc commit offset là điều cốt yếu khi làm việc với Kafka. Theo mặc định, message sẽ được tự động commit sau một khoảng thời gian nhất định. Để biết thêm thông tin, xem tài liệu KafkaJS. KafkaContext cung cấp cách để truy cập consumer đang hoạt động nhằm commit offset thủ công. Consumer chính là consumer của KafkaJS và hoạt động như cách triển khai gốc của KafkaJS.

@EventPattern('user.created')
async handleUserCreated(@Payload() data: IncomingMessage, @Ctx() context: KafkaContext) {
  // logic nghiệp vụ

  const { offset } = context.getMessage();
  const partition = context.getPartition();
  const topic = context.getTopic();
  const consumer = context.getConsumer();
  await consumer.commitOffsets([{ topic, partition, offset }])
}

@Bind(Payload(), Ctx())
@EventPattern('user.created')
async handleUserCreated(data, context) {
  // logic nghiệp vụ

  const { offset } = context.getMessage();
  const partition = context.getPartition();
  const topic = context.getTopic();
  const consumer = context.getConsumer();
  await consumer.commitOffsets([{ topic, partition, offset }])
}

Để tắt tính năng tự động commit message, đặt autoCommit: false trong cấu hình run, như sau:

const app = await NestFactory.createMicroservice<MicroserviceOptions>(AppModule, {
  transport: Transport.KAFKA,
  options: {
    client: {
      brokers: ['localhost:9092'],
    },
    run: {
      autoCommit: false
    }
  }
});

const app = await NestFactory.createMicroservice(AppModule, {
  transport: Transport.KAFKA,
  options: {
    client: {
      brokers: ['localhost:9092'],
    },
    run: {
      autoCommit: false
    }
  }
});

Cập nhật trạng thái instance

Để nhận cập nhật thời gian thực về kết nối và trạng thái của driver instance bên dưới, bạn có thể subscribe vào stream status. Stream này cung cấp các cập nhật trạng thái đặc thù cho driver đã chọn. Với driver Kafka, stream status phát ra các event connected, disconnected, rebalancing, crashed, và stopped.

this.client.status.subscribe((status: KafkaStatus) => {
  console.log(status);
});

Tương tự, bạn có thể subscribe vào stream status của server để nhận thông báo về trạng thái của server.

const server = app.connectMicroservice<MicroserviceOptions>(...);
server.status.subscribe((status: KafkaStatus) => {
  console.log(status);
});

Truy cập producer và consumer bên dưới

Đối với các use case phức tạp hơn, bạn có thể cần truy cập trực tiếp instance producer và consumer bên dưới. Điều này có thể hữu ích cho các tình huống như tự đóng kết nối thủ công hoặc dùng các phương thức đặc thù của driver. Tuy nhiên, hãy nhớ rằng trong hầu hết trường hợp, bạn không nên cần truy cập trực tiếp driver.

Để làm điều này, bạn có thể dùng getter producer và consumer được expose bởi instance ClientKafkaProxy.

const producer = this.client.producer;
const consumer = this.client.consumer;