kafka管控推荐使用 滴滴开源 的 Kafka运维管控平台 更符合国人的操作习惯 ,

更强大的管控能力 ,更高效的问题定位能力 、更便捷的集群运维能力 、更专业的资源治理 、 更友好的运维生态

前言

我们会看到zk的数据中有一个节点/log_dir_event_notification/，这是一个序列号持久节点
这个节点在kafka中承担的作用是: 当某个Broker上的LogDir出现异常时(比如磁盘损坏,文件读写失败,等等异常): 向zk中谢增一个子节点/log_dir_event_notification/log_dir_event_序列号 ；Controller监听到这个节点的变更之后,会向Brokers们发送LeaderAndIsrRequest请求; 然后做一些副本脱机的善后操作

源码分析

这里说的dirLog是 server.properties中配置的log.dir 例如
在这里插入图片描述

副本异常处理

首先我们找到有使用这个节点的源码;
kafka启动之初有调用

ReplicaManager.startup()

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
def startup(): Unit = {
  // 省略...
  
  //当inter-broker protocol (IBP) < 1.0的时候,如果存在logDir的一些异常则直接让整个Broker启动失败;
  val haltBrokerOnFailure = config.interBrokerProtocolVersion < KAFKA_1_0_IV0
  logDirFailureHandler = new LogDirFailureHandler("LogDirFailureHandler", haltBrokerOnFailure)
  logDirFailureHandler.start()
}
private class LogDirFailureHandler(name: String, haltBrokerOnDirFailure: Boolean) extends ShutdownableThread(name) {
  override def doWork(): Unit = {
    //从队列 offlineLogDirQueue 取数据
    val newOfflineLogDir = logDirFailureChannel.takeNextOfflineLogDir()
    if (haltBrokerOnDirFailure) {
      fatal(s"Halting broker because dir $newOfflineLogDir is offline")
      Exit.halt(1)
    }
    handleLogDirFailure(newOfflineLogDir)
  }
}
1
2
3
4
5
6
7
8
9
10
11
// logDir should be an absolute path
  // sendZkNotification is needed for unit test
  def handleLogDirFailure(dir: String, sendZkNotification: Boolean = true): Unit = {
       // 省略...
    logManager.handleLogDirFailure(dir)

    if (sendZkNotification)
      zkClient.propagateLogDirEvent(localBrokerId)
    warn(s"Stopped serving replicas in dir $dir")
  }

代码比较长,就直接概况一下好了：
主要是当读取或操作LogDir的时候出现了异常就会执行到这里,有可能是磁盘脱机了,或者文件突然没有读取写入权限等等之类的一些IOException异常;那么 Broker就需要做一些处理；如下

  1. 做个判断inter.broker.protocol.version 协议版本 < 1.0 的时候 时候直接退出；那个时候还不支持单Broker上存在多个logDir;
  2. 副本停止fetche数据
  3. 标记分区下线
  4. 可能移除一些监控信息
  5. 如果当前的log_dir 都脱机(或者异常了), 那么久可以直接shutdown这台机器了
  6. 如果还有其他的log_dir 还有在线的, 那么继续做一些其他的清理操作;
  7. 创建持久序列节点/log_dir_event_notification/log_dir_event_+序列号；数据是 BrokerID;例如:
    /log_dir_event_notification/log_dir_event_0000000003 
    1
    {"version":1,"broker":20003,"event":1}

PS： log_dir 是可以在一台Broker配置多个路径的 ，用逗号隔开

LogDir发生异常

比如说在 给文件加锁的时候lockLogDirs,磁盘损坏了就抛出异常IOException

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
/**
 * Lock all the given directories
 */
private def lockLogDirs(dirs: Seq[File]): Seq[FileLock] = {
  dirs.flatMap { dir =>
    try {
      val lock = new FileLock(new File(dir, LockFile))
      if (!lock.tryLock())
        throw new KafkaException("Failed to acquire lock on file .lock in " + lock.file.getParent +
          ". A Kafka instance in another process or thread is using this directory.")
      Some(lock)
    } catch {
      case e: IOException =>
        logDirFailureChannel.maybeAddOfflineLogDir(dir.getAbsolutePath, s"Disk error while locking directory $dir", e)
        None
    }
  }
}
def maybeAddOfflineLogDir(logDir: String, msg: => String, e: IOException): Unit = {
  error(msg, e)
  if (offlineLogDirs.putIfAbsent(logDir, logDir) == null)
    offlineLogDirQueue.add(logDir)
}

offlineLogDirQueue添加了一个异常队列之后就回到上面的副本异常处理代码了, 上面可是一致在queue.take()的

Controller监听zk节点变更

KafkaController.processLogDirEventNotification

1
2
3
4
5
6
7
8
9
10
11
12
private def processLogDirEventNotification(): Unit = {
  if (!isActive) return
  val sequenceNumbers = zkClient.getAllLogDirEventNotifications
  try {
    val brokerIds = zkClient.getBrokerIdsFromLogDirEvents(sequenceNumbers)
    //尝试将这台Broker上的所有副本 走一下状态流转 到 OnlineReplica
    onBrokerLogDirFailure(brokerIds)
  } finally {
    // delete processed children
    zkClient.deleteLogDirEventNotifications(sequenceNumbers, controllerContext.epochZkVersion)
  }
}

主要将从zk节点 /log_dir_event_notification/log_dir_event_序列号 中获取到的数据的Broker上的所有副本进行一个副本状态流转 ->OnlineReplica ；关于状态机的流转请看 【kafka源码】Controller中的状态机

  1. 给所有broker 发送LeaderAndIsrRequest请求，让brokers们去查询他们的副本的状态，如果副本logDir已经离线则返回KAFKA_STORAGE_ERROR异常;
  2. 完事之后会删除节点

源码总结

Q&A

作者石臻臻,工作8年的互联网老兵,丰富的开发和管理经验,全网「 粉丝数4万 」,

先后从事 「 电商 」、「 中间件 」、「 大数据」 等工作

现在任职于「 滴滴技术专家 」岗位,从事开源建设工作

目前在维护 个人公众号「  石臻臻的杂货铺 」 ; 关注公众号会有「 日常送书活动 」;

欢迎进「 高质量 」 「 滴滴开源技术答疑群 」 , 群内每周技术专家轮流值班答疑

可帮忙「 内推 」一二线大厂