java kafka无法消费,Kafka异常重启后无法消费

java kafka无法消费,Kafka异常重启后无法消费

事故起因

2018.10.13日Kafka出现了无法启动问题，安稳了一天，结果第二天晚上又出现了新的问题，发现有些程序无法正常消费Kafka了。

这个问题在网上找了一下，发现有类似的经历，都是Kafka进程异常挂掉之后，起来无法消费的问题，但是试了一下，发现每次事故还是都有它的独特之处，所以这里专门记录一下。

调查过程

基本环境信息：

软件版本：使用的Kakfa的版本是1.1.0版本，之前因为从0.8.2升级过一次，所以消息格式仍然是沿用支持0.8.2的消息格式。

集群：6个节点，broker id分别是：0，1，2，3，4，5

节点配置：内存：64G 硬盘：4T * 12

网卡：万兆

其中0，1两个节点的Kafka出现过异常挂掉。

出现问题的时候，也不是所有的程序都无法消费，而是部分节点无法消费了，查了一下机器的负载情况，0号节点负载非常高，达到11了，磁盘利用率非常高，而其他的几个节点的负载都非常低。

这里我们简单重温一下消费的原理：

同一个组的消费者会加入到一起，向Coordinator请求加入消费组，Coordinator会向每个消费者分配消费的分区。

那么Coordinator是怎么确定的呢？一般一个消费者启动后，会随机想一个节点请求去获取它的Coordinator所在节点。

获取Coordinator所在的节点的原理与我们的消费组的消费偏移量信息是存储__consumer_offsets对应的分区上。

所以最终定位到Kafka是如何选择__consumer_offsets这个topic的分区来存储对应的偏移量。

以消费组名为ttt为例来说明：

int hashCode = Math.abs("ttt".hashCode());

int partition = hashCode % 50;

先计算group的hashCode，再除以分区数(50),可以得到partition的值为：44。

再通过kafka-topic.sh命令查看该partition的leader在哪个节点，那么就可以确定tttt这个消费组的Coordinator落在哪个节点了。

在消费过程中，消费不了的原因是说对应的Coordinator无效，日志如下：

[2018-10-14 21:20:28,860] INFO [Consumer clientId=consumer-1, groupId=console-consumer-28334] Discovered group coordinator wxoddc2nn1:9092 (id: 2147483647 rack: null) (org.apache.sumer.internals.AbstractCoordinator)

[2018-10-14