运行两个不同主题的Kafka消费者时遭遇CommitFailedException求助

首先咱们得揪出核心问题：你的两个消费者程序用了完全相同的group.id值test。Kafka的消费者组机制是，同一个组内的消费者会协同消费订阅的主题分区——当你启动第二个同组消费者时，Kafka会触发组重平衡，重新分配分区给组内成员。在重平衡过程中，旧成员（第一个消费者）的会话会被标记为失效，这时候它再尝试提交偏移量，就会抛出你看到的CommitFailedException。

接下来给你几个具体的解决步骤，按优先级排序：

1. 给两个消费者分配不同的group.id #

这是最直接有效的方案，因为你的两个消费者分别订阅不同的主题（Test1和Test2），完全不需要属于同一个消费组。修改两个消费者代码里的group.id配置：

• 订阅Test1的消费者：props.put("group.id", "test-group-1");
• 订阅Test2的消费者：props.put("group.id", "test-group-2");

这样两个消费组完全独立，各自处理自己的主题，不会触发相互之间的重平衡，自然就不会出现提交失败的问题。

2. 优化偏移量提交逻辑 #

看你的代码，每次处理一条记录就调用consumer.commitSync()，这不仅效率低，还会增加提交失败的概率。应该在处理完整一批poll到的记录后再提交偏移量，把提交操作移到for循环外面：

try {
    while (true) {
        ConsumerRecords<String, String> records = consumer.poll(100);
        if (!records.isEmpty()) {
            for (ConsumerRecord<String, String> record : records) {
                System.out.println("Record: "+record.value());
                // 这里执行你的业务处理逻辑
            }
            // 处理完当前批次所有记录后统一提交
            consumer.commitSync();
        }
    }
} catch (Exception e) {
    LOG.error("Exception: ", e);
} finally {
    consumer.close();
}

这样减少了提交次数，也避免了在处理过程中因重平衡导致的提交冲突。

3. 验证参数配置的合理性 #

虽然你已经调整了session.timeout.ms和max.poll.records，但还是要注意几个参数的关联逻辑：

• heartbeat.interval.ms应该是session.timeout.ms的1/3左右，你设置的1000ms对应30000ms的session超时，这个比例是合理的。
• 如果你的业务逻辑处理单条记录的时间仍然很长，可以进一步减小max.poll.records，或者把session.timeout.ms调至60000ms，确保在session超时前能完成当前批次的处理并调用下一次poll。

最后提醒：如果确实需要让两个消费者属于同一个组（比如要消费多主题分区并分摊负载），那要确保它们订阅的主题集合完全一致，否则重平衡会频繁发生，也容易引发类似异常。

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