Kafka:Leader 丢失导致的 Consumer 挂起故障解决

栏目: 后端 · 发布时间: 6年前

内容简介:最近发现线上的Kafka Consumer Client频繁出现无法消费的情况,导致offset积压。但是在重启Kafka Broker之后又正常了。 而Cloudera Manager在重启之前,我们发现三台broker中并没有KakfaController。让人很是不解。目前已经没办法复现当时的场景,我们简单描述下,通过命令检查Topic的状态如下:

现象

最近发现线上的Kafka Consumer Client频繁出现无法消费的情况,导致offset积压。但是在重启Kafka Broker之后又正常了。 而Cloudera Manager在重启之前,我们发现三台broker中并没有KakfaController。让人很是不解。

排查步骤

检查Topic的状态

目前已经没办法复现当时的场景,我们简单描述下,通过命令

kafka-topics --zookeeper hadoop02 --desc

检查Topic的状态如下:

Kafka:Leader 丢失导致的 Consumer 挂起故障解决

这张图是后续补的,当时的现象是Leader的值均为-1,Isr的值也均为-1

至此我们猜测是因为KakfaController丢失导致的partition leader为-1,进而导致的Consumer端无法正常消费。

查看Leader的选举方式

在这里我们需要先看下KafkaController,KafkaServer.startup()的时候会新建KafkaController,而KafkaController在启动时,启动了Controller的Elector

def startup() = {
  inLock(controllerContext.controllerLock) {
    info("Controller starting up");
    registerSessionExpirationListener()
    isRunning = true
    controllerElector.startup
    info("Controller startup complete")
  }
}

我们再点进去查看下Controller是怎么elect出来的:

def elect: Boolean = {
  val timestamp = SystemTime.milliseconds.toString
  val electString = Json.encode(Map("version" -> 1, "brokerid" -> brokerId, "timestamp" -> timestamp))
 
 leaderId = getControllerID 
  /* 
   * We can get here during the initial startup and the handleDeleted ZK callback. Because of the potential race condition, 
   * it's possible that the controller has already been elected when we get here. This check will prevent the following 
   * createEphemeralPath method from getting into an infinite loop if this broker is already the controller.
   */
  if(leaderId != -1) {
     debug("Broker %d has been elected as leader, so stopping the election process.".format(leaderId))
     return amILeader
  }

  try {
    createEphemeralPathExpectConflictHandleZKBug(controllerContext.zkClient, electionPath, electString, brokerId,
      (controllerString : String, leaderId : Any) => KafkaController.parseControllerId(controllerString) == leaderId.asInstanceOf[Int],
      controllerContext.zkSessionTimeout)
    info(brokerId + " successfully elected as leader")
    leaderId = brokerId
    onBecomingLeader()
  } catch {
    case e: ZkNodeExistsException =>
      // If someone else has written the path, then
      leaderId = getControllerID 

      if (leaderId != -1)
        debug("Broker %d was elected as leader instead of broker %d".format(leaderId, brokerId))
      else
        warn("A leader has been elected but just resigned, this will result in another round of election")

    case e2: Throwable =>
      error("Error while electing or becoming leader on broker %d".format(brokerId), e2)
      resign()
  }
  amILeader
}

从上面的选举代码中我们可以看出 在Kafka集群刚启动时,默认Broker谁先启动则默认为Controller,后续若/controller节点发生变化,会触发Leader变更监听程序LeaderChangeListener,执行变更操作或者重新选举

inLock(controllerContext.controllerLock) {
  debug("%s leader change listener fired for path %s to handle data deleted: trying to elect as a leader"
    .format(brokerId, dataPath))
  if(amILeader)
    onResigningAsLeader()
  elect
}

我们知道kafka消费的时候需要和Leader通信,而Leader不存在导致的没办法消费很容易理解,那么为什么Controller丢失会导致partition的leader不正常呢?我们来看下面一张图片(图片来自CSDN博主:happy19870612):

Kafka:Leader 丢失导致的 Consumer 挂起故障解决

val replicaStateMachine = new ReplicaStateMachine(this)

我们可以发现Replica的状态机管理是在KafkaController中完成的,也就是说Controller丢失的情况下,也就失去了与Zookeeper交互的能力。默认情况下Leader必须从ISR列表中选择,我们发现列表为空(经过排查发现是Kafka的bug,在Controller和Zookeeper通信过程中出现问题时,可能导致leader丢失而无法通信的情况,这个可能性是很大的,因为zookeeper在高并发环境是容易超时,这就是为什么在kafka 0.8.2.1之后更建议我们使用kafka topic的方式存储offset而不是存储在zookeeper中。

检查系统日志

当然一般情况下我们会先检查系统日志是否有报异常,这种定位问题效率最高。我们来看看kafka的server log是不是有和zookeeper相关的异常

Kafka:Leader 丢失导致的 Consumer 挂起故障解决

从图上可以发现,的确存在zookeeper连接失败的情况,另外我们发现一个比较诡异的事情:

2018-04-10 00:30:11,149 INFO kafka.controller.KafkaController: [Controller 218]: Currently active brokers in the cluster: Set()

所有的broker都临时下线了,然后我查看了其他broker的server log发现所有机器在同一时间均出现了zookeeper连接超时的情况,导致了后续一连串的ERROR:

Kafka:Leader 丢失导致的 Consumer 挂起故障解决

解决方式

那么问题已经很明了了,我们又检查了下凌晨的网络IO:

Kafka:Leader 丢失导致的 Consumer 挂起故障解决

发现凌晨的时候出现了流量高峰。可见主要原因还是当时系统IO压力大导致的连接超时,于是我们适当调大了zookeeper.session.timeout.ms=12,再也没有出现过超时的情况。实际上这是kafka的bug,在0.10版本中已经解决了。另一方面在我们集群中出现这个问题,有一部分原因在于,我们的kafka集群和zookeeper集群是在不同的机房,不同的网络中。在后来的排查中我们发现有一台交换机网口有问题导致的带宽不稳定,特别是在大规模计算的时候尤为明显。

PS:当zookeeper服务器端和客户端版本不一致的时候也会导致连接超时的情况。


以上所述就是小编给大家介绍的《Kafka:Leader 丢失导致的 Consumer 挂起故障解决》,希望对大家有所帮助,如果大家有任何疑问请给我留言,小编会及时回复大家的。在此也非常感谢大家对 码农网 的支持!

查看所有标签

猜你喜欢:

本站部分资源来源于网络,本站转载出于传递更多信息之目的,版权归原作者或者来源机构所有,如转载稿涉及版权问题,请联系我们

C算法(第二卷:图算法)(第3版)

C算法(第二卷:图算法)(第3版)

塞德威克(Sedgewick Robert) / 周良忠 / 第1版 (2004年1月1日) / 2004-4 / 38.0

《C算法(第2卷)(图算法)(第3版)(中文版)》所讨论的图算法,都是实际中解决图问题的最重要的已知方法。《C算法(第2卷)(图算法)(第3版)(中文版)》的主要宗旨是让越来越多需要了解这些算法的人的能够掌握这些方法及基本原理。书中根据基本原理从基本住处开始循序渐进地讲解,然后再介绍一些经典方法,最后介绍仍在进行研究和发展的现代技术。精心挑选的实例、详尽的图示以及完整的实现代码与正文中的算法和应用......一起来看看 《C算法(第二卷:图算法)(第3版)》 这本书的介绍吧!

SHA 加密
SHA 加密

SHA 加密工具

正则表达式在线测试
正则表达式在线测试

正则表达式在线测试

HEX HSV 转换工具
HEX HSV 转换工具

HEX HSV 互换工具