如何使用NestJS和Jest对Kafka消息生产/消费进行E2E测试

核心问题原因 #

你遇到的spyOn失效，本质是NestJS的HTTP控制器实例与Kafka消费者控制器实例并非同一个对象：

• 通过app.get(Controller)获取的是HTTP模块初始化的控制器实例
• Kafka消费者的事件处理逻辑，是由NestJS微服务模块单独创建的另一个控制器实例执行的
• 因此你绑定到HTTP实例上的spy无法捕获到微服务消费者的调用

同时，消息生产消费是异步流程，固定setTimeout的断言时机不可靠，容易导致测试不稳定。

解决方案1：直接验证消息生产（推荐） #

跳过对消费端的spy，直接拦截Kafka生产者的send方法，验证消息是否正确发送到目标主题：

import { Kafka } from '@nestjs/microservices';

test('Should produce message to correct topic', async () => {
  // 获取Kafka生产者实例并创建spy
  const kafkaProducer = app.get<Kafka>(Kafka).producer();
  const sendSpy = jest.spyOn(kafkaProducer, 'send').mockResolvedValue({});

  // 触发生产消息的接口
  await request(app.getHttpServer())
    .get('/')
    .expect(200)
    .expect('Hello World!');

  // 验证生产者调用参数
  expect(sendSpy).toHaveBeenCalledTimes(1);
  expect(sendSpy).toHaveBeenCalledWith(expect.objectContaining({
    topic: 'queuing.restaurant.ordered_hotdogs',
    messages: expect.arrayContaining([
      expect.objectContaining({
        value: expect.any(Buffer) // 对应Schema Registry编码后的消息
      })
    ])
  }));

  // 恢复原方法（可选，避免影响其他测试）
  sendSpy.mockRestore();
});

解决方案2：端到端验证消费流程 #

如果需要完整验证生产+消费链路，可以创建一个测试专用的消费者，主动订阅目标主题并接收消息：

import { Kafka } from '@nestjs/microservices';

test('Should produce and consume message correctly', async () => {
  // 1. 初始化测试消费者
  const testConsumer = app.get<Kafka>(Kafka).consumer({ groupId: 'test-e2e-group' });
  await testConsumer.connect();
  await testConsumer.subscribe({ 
    topic: 'queuing.restaurant.ordered_hotdogs', 
    fromBeginning: true 
  });

  let receivedMessage: any;
  // 启动消费者监听消息
  await testConsumer.run({
    eachMessage: async ({ message }) => {
      // 复用业务代码的解码逻辑
      const id = await registry.getLatestSchemaId(KafkaTopics.ORDERED_HOTDOGS_SUBJECT);
      const schema = await registry.getSchema(id);
      receivedMessage = await registry.decode(message.value);
    },
  });

  // 2. 触发消息生产
  await request(app.getHttpServer())
    .get('/')
    .expect(200)
    .expect('Hello World!');

  // 3. 轮询等待消息被消费（比固定timeout更可靠）
  await new Promise((resolve) => {
    const checkInterval = setInterval(() => {
      if (receivedMessage) {
        clearInterval(checkInterval);
        resolve(null);
      }
    }, 100);
  }, 10000); // 设置超时时间，避免无限等待

  // 4. 验证消费的消息内容
  const expectedMsg = { id: "1", name: "Hotdog", topping: "pickles" };
  expect(receivedMessage).toEqual(expectedMsg);

  // 5. 清理测试资源
  await testConsumer.disconnect();
});

额外优化建议 #

• 测试前可以调用Kafka的admin API清空目标主题，避免历史消息干扰测试结果
• 将Schema Registry的解码逻辑封装成公共工具函数，业务代码和测试代码复用，避免重复实现导致的不一致

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