本地环境下KafkaJS消费者无法消费消息问题排查求助

问题描述 #

本地环境使用KafkaJS实现的消费者进程显示处于运行状态，但无法接收生产者持续推送的主题事件。该消费者一周前可正常工作，当前相同配置在开发环境中能正常消费处理事件，本地环境异常原因无法定位。

代码实现 #

Kafka初始化与消费者工具函数：

const kafka = new Kafka({
  clientId: "integrations-queues",
  brokers: [
    process.env.KAKFA_CONNECTION_URL_1,
    process.env.KAKFA_CONNECTION_URL_2,
  ],
});

const createKafkaConsumer = (groupId: string) => {
  const consumer = kafka.consumer({
    groupId,
  });

  return consumer;
};

const connectAndSubscribeKafkaConsumer = async (
  consumer: Consumer,
  topic: string
) => {
  try {
    await consumer.connect();

    await consumer.subscribe({ topic });
  } catch (error) {
    SentryService.captureException(error);
    console.log(error);
  }
};

消费者启动逻辑：

(async () => {
  try {
    const consumerGroupName =
      process.env.ENV === "localhost"
        ? "raw-data-consumer-local"
        : process.env.WORKFLOW_RAW_DATA_CONSUMER;

    rawDataConsumer = createKafkaConsumer(consumerGroupName);

    await connectAndSubscribeKafkaConsumer(
      rawDataConsumer,
      process.env.WORKFLOW_EXECUTIONS_PRODUCER_TOPIC
    );

    await rawDataConsumer.run({
      eachMessage: rawEventHandler,
    });
  } catch (error) {
    SentryService.captureException(error);
    console.log(error);
  }
})();

排查与解决方案 #

1. 校验消费者组偏移量 #

用Kafka命令行工具查看本地消费者组的偏移量状态：

kafka-consumer-groups.sh --bootstrap-server <你的Broker地址> --describe --group raw-data-consumer-local

• 若CURRENT-OFFSET等于LOG-END-OFFSET：说明消费者已追上最新偏移量，需确认生产者是否真的向本地环境对应的topic推送了消息；或尝试重置偏移量到最早位置测试。
• 若CURRENT-OFFSET显示为-：说明消费者组未分配到分区，可能是topic不存在、权限不足或Broker连接异常。

2. 修正订阅逻辑的偏移量策略 #

当前订阅未指定fromBeginning，默认从最新偏移量开始消费。如果本地消费者组之前已消费到最新位置，新消息未被消费可能是偏移量问题，临时修改订阅代码测试：

await consumer.subscribe({ topic, fromBeginning: true });

3. 排查消息处理函数的静默异常 #

检查rawEventHandler是否存在内部吞异常的情况，添加日志输出确认是否接收到消息：

const rawEventHandler = async ({ topic, partition, message }) => {
  console.log(`[DEBUG] 收到消息：topic=${topic}, partition=${partition}, value=${message.value?.toString()}`);
  // 原有业务逻辑
};

4. 验证本地环境的连接与权限 #

• 确认本地环境的KAKFA_CONNECTION_URL_1/KAKFA_CONNECTION_URL_2配置与开发环境一致，且能正常访问Broker地址。
• 检查本地消费者是否有目标topic的消费权限：

kafka-acls.sh --bootstrap-server <Broker地址> --list --topic ${process.env.WORKFLOW_EXECUTIONS_PRODUCER_TOPIC} --group raw-data-consumer-local

5. 开启KafkaJS调试日志 #

初始化Kafka时开启debug日志，排查连接、订阅、分区分配的细节：

const { logLevel } = require('kafkajs');
const kafka = new Kafka({
  clientId: "integrations-queues",
  brokers: [process.env.KAKFA_CONNECTION_URL_1, process.env.KAKFA_CONNECTION_URL_2],
  logLevel: logLevel.DEBUG
});

6. 重置消费者组偏移量（可选） #

若确认偏移量异常，可重置偏移量到最早位置：

kafka-consumer-groups.sh --bootstrap-server <Broker地址> --reset-offsets --group raw-data-consumer-local --topic ${process.env.WORKFLOW_EXECUTIONS_PRODUCER_TOPIC} --to-earliest --execute

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