如何编写NestJS Kafka端到端测试？官方示例仓库未找到相关内容

核心思路 #

用Testcontainers启动临时Kafka容器，模拟真实Kafka环境，避免依赖外部服务，同时通过直接操作Kafka客户端验证生产/消费逻辑的正确性。

1. 依赖准备 #

• 安装必要依赖包：

  npm install --save-dev testcontainers kafkajs @nestjs/testing
  

• 确保项目已集成@nestjs/microservices和Kafka客户端模块。

2. 测试环境初始化 #

在E2E测试文件中，先启动Kafka容器，再初始化Nest测试模块，注入生产/消费服务：

import { Test } from '@nestjs/testing';
import { KafkaContainer, StartedKafkaContainer } from 'testcontainers';
import { Kafka } from 'kafkajs';
import { KafkaProducerService } from '../src/producer/kafka-producer.service';
import { KafkaConsumerService } from '../src/consumer/kafka-consumer.service';

describe('Kafka E2E Test Suite', () => {
  let kafkaContainer: StartedKafkaContainer;
  let producerService: KafkaProducerService;
  let consumerService: KafkaConsumerService;
  let kafkaClient: Kafka;

  beforeAll(async () => {
    // 启动Kafka容器，获取Broker地址
    kafkaContainer = await new KafkaContainer().start();
    const bootstrapServers = kafkaContainer.getBootstrapServers();

    // 创建测试模块，注入自定义服务和Kafka配置
    const moduleRef = await Test.createTestingModule({
      providers: [
        KafkaProducerService,
        KafkaConsumerService,
        {
          provide: 'KAFKA_BOOTSTRAP_SERVERS',
          useValue: bootstrapServers,
        },
      ],
    }).compile();

    producerService = moduleRef.get(KafkaProducerService);
    consumerService = moduleRef.get(KafkaConsumerService);

    // 初始化独立Kafka客户端，用于直接验证消息
    kafkaClient = new Kafka({ brokers: [bootstrapServers] });
  });

  afterAll(async () => {
    // 清理资源
    await kafkaContainer.stop();
    await kafkaClient.disconnect();
  });
});

3. 测试生产者逻辑 #

验证生产者能成功发送消息到指定主题：

it('should send message to Kafka topic successfully', async () => {
  const testTopic = 'test-user-events';
  const testPayload = { userId: 123, action: 'created' };

  // 创建临时消费者监听目标主题
  const consumer = kafkaClient.consumer({ groupId: 'e2e-test-group' });
  await consumer.subscribe({ topic: testTopic, fromBeginning: true });

  let receivedPayload: any;
  // 启动消费者，捕获收到的消息
  await consumer.run({
    eachMessage: async ({ message }) => {
      receivedPayload = JSON.parse(message.value.toString());
    },
  });

  // 调用生产者服务发送消息
  await producerService.send(testTopic, testPayload);

  // 等待Kafka消息传递（适配异步时序）
  await new Promise(resolve => setTimeout(resolve, 1000));

  // 断言消息内容一致
  expect(receivedPayload).toEqual(testPayload);
  await consumer.disconnect();
});

4. 测试消费者逻辑 #

验证消费者能正确接收并处理消息：

it('should consume and process Kafka message correctly', async () => {
  const testTopic = 'test-user-events';
  const testPayload = { userId: 456, action: 'updated' };

  // 用独立生产者发送测试消息
  const producer = kafkaClient.producer();
  await producer.connect();
  await producer.send({
    topic: testTopic,
    messages: [{ value: JSON.stringify(testPayload) }],
  });
  await producer.disconnect();

  // 等待消费者处理消息
  await new Promise(resolve => setTimeout(resolve, 1000));

  // 验证处理结果（示例：假设消费者会把处理后的消息存到服务内部状态）
  const lastProcessed = await consumerService.getLastProcessedMessage();
  expect(lastProcessed).toEqual(testPayload);
});

5. 关键注意事项 #

• 异步时序处理：Kafka消息传递是异步的，测试中需设置合理延迟或使用Promise等待，避免断言过早执行。
• 资源清理：每个测试后重置主题或停止容器，防止脏数据影响后续测试。
• CI环境适配：可通过环境变量控制是否启用Testcontainers，比如在CI中使用预配置的Kafka服务。

内容的提问来源于stack exchange，提问作者mh377