一次KAFKA消费者异常引发的思考

问题描述：

线上出现一台服务器特别慢，因而关闭了服务器上的kafka broker. 关闭后发现一些kafka consumer没法正常消费数据了, 日志错误：
o.a.kakfa.clients.consumer.internals.AbstractCordinator Marking the coordinator (39.0.2.100) as dead.

缘由：

通过一番排查，发现consumer group信息:
(kafka.coordinator.GroupMetadataMessageFormatter类型):
groupId::[groupId,Some(consumer),groupState,Map(memberId -> [memberId,clientId,clientHost,sessionTimeoutMs], ...->[]...)],
存到了KAFKA内部topic: __consumer_offsets里, , 它的key是 groupId.
同时发现broker 参数 offsets.topic.replication.factor 错误地被设置为1. 这个参数表示TOPIC: __Consumer_offsets 的副本数. 这样一旦某个broker被关闭, 若是被关闭的Broker 是__Consumer_offsets的某些partition的Leader. 则致使某些consumer group 不可用. 若是一旦broker已经启动, 须要手工经过命令行来扩展副本数.

reassignment.json:
{"version":1,
 "partitions": [{"topic": "xxx", "partition": 0, "replicas": {brokerId1, brokerId2}}]
}
kafka-reassign-partitions  --zookeeper localhost:2818 --reassignment-json-file  reassignment.json --execute

客户端寻找Consumer Coordinator的过程:
客户端 org.apache.kafka.clients.consumer.internals.AbstractCoordinator
若是Coordinator 未知 (AbstractCoordinator.coordinatorUnknown()), 发起请求 lookupCoordinator，向负载最低的节点发送FindCoordinatorRequest

服务端 KafkaApis.handleFindCoordinatorRequest 接收请求：
首先调用 GroupMetaManager.partitionFor(consumerGroupId) consunerGroupId 的 hashCode 对 __consumer_offsets 总的分片数取模获取partition id 再从 __consumer_offset 这个Topic 中找到partition对应的 Partition Metadata, 而且获取对应的Partition leader 返回给客户端

引申思考

KAFKA 的failover机制到底是怎么样的？假使 __consumer_offset 设置了正确的副本数，重选举的过程是怎样的. 若是broker宕机后致使某些副本不可用, 副本会自动迁移到其余节点吗？带着这些问题稍微阅读了一下KAFKA的相关代码:

当一个Broker 被关掉时, 会有两步操做：
KafkaController.onBrokerFailure ->KafkaController.onReplicasBecomeOffline
主要是经过 PartitionStateMachine.handleStateChanges 方法通知Partition状态机将状态置为offline. ReplicaStateMachine.handleStateChanges方法会将Replica 状态修改成OfflineReplica, 同时修改partition ISR. 若是被关闭broker 是partition leader 那么须要从新触发partition leader 选举，最后发送LeaderAndIsrRequest获取最新的Leader ISR 信息.
KafkaController.unregisterBrokerModificationsHandler 取消注册的BrokerModificationsHandler 并取消zookeeper 中broker 事件的监听.

当ISR请求被发出,KafkaApis.handleLeaderAndIsrRequest() 会被调用. 这里若是须要变动leader的partition是属于__consumer_offset这个特殊的topic,取决于当前的broker节点是否是partition leader. 会分别调用GroupCoordinator.handleGroupImmigration 和 GroupCoordinator.handleGroupEmmigration. 若是是partition leader, GroupCoordinator.handleGroupImmigration -> GroupMetadataManager.loadGroupsForPartition 会从新从 __consumer_offset 读取group数据到本地metadata cache, 若是是partition follower, GroupCoordniator.handleGroupImigration -> GroupMetadataManager.removeGroupsForPartition 会从metadata cache中移除group信息. 并在onGroupUnloaded回调函数中将group的状态变动为dead. 同时通知全部等待join或者sync的组成员.

KAFKA在Broker关闭时不会自动作partition 副本的迁移. 这时被关闭的Broker上的副本变为under replicated 状态. 这种状态将持续直到Broker被从新拉起而且追上新的数据, 或者用户经过命令行 手动复制副本到其余节点.

官方建议设置两个参数来保证graceful shutdown. controlled.shutdown.enable=true auto.leader.rebalance.enable=true前者保证关机以前将日志数据同步到磁盘，并进行重选举. 后者保证在broker从新恢复后再次得到宕机前leader状态. 避免leader分配不均匀致使读写热点.

Reference

https://blog.csdn.net/zhanglh046/article/details/72833129
https://blog.csdn.net/huochen1994/article/details/80511038
https://www.jianshu.com/p/1aba6e226763