KafkaConsumer 被 8 个线程共用 报 ConcurrentModificationException 消费停摆

while (running) {
  ConsumerRecords<String,String> rs = consumer.poll(Duration.ofMillis(300));
  executor.submit(() -> handle(rs));
}
// 提交仍在 poll 线程

为了提速把同一个 KafkaConsumer 实例传给 8 个工作线程，跑几分钟后抛 ConcurrentModificationException，随后 poll 一直返回空，lag 持续上涨。这类现象通常不是突然出现的，往前翻半天日志能看到苗头。

追下去是这么回事：KafkaConsumer 不是线程安全的，内部订阅状态和缓冲在并发访问下被破坏。官方只允许一个线程调用 poll，其他线程连 wakeup 都要谨慎。

「KafkaConsumer 线程安全」改完先别急着全量，挑一台机器放半天，真有问题这段时间就能看出来。监控和日志的时间要对上，时区不一致会多绕很大一圈。

临时改动用注释写上日期和原因，过两周自己都记不清为什么这么写。动手前把原文件复制一份带时间戳，出问题立刻换回去，比现场分析快得多。

处理上按这个顺序：改成一个线程负责 poll 把记录投给业务线程池；或者每个线程持有独立 consumer 并手动 assign 分区；提交位点收回到 poll 那个线程统一做；重启后确认 lag 恢复下降。

确认没问题再收工，压测十分钟不再出现 ConcurrentModificationException 八个线程都在处理记录。