本篇教程探讨了大数据技术之一次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分配不均匀导致读写热点.

本文由职坐标整理发布，学习更多的相关知识，请关注职坐标IT知识库！