【kafka源码】/log_dir_event_notification的LogDir脱机事件通知

前言

我们会看到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()

  def startup(): Unit = {// 省略...//当inter-broker protocol (IBP) < 1.0的时候,如果存在logDir的一些异常则直接让整个Broker启动失败;val haltBrokerOnFailure = config.interBrokerProtocolVersion < KAFKA_1_0_IV0logDirFailureHandler = 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)}}

// logDir should be an absolute path// sendZkNotification is needed for unit testdef 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就需要做一些处理；如下

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

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

LogDir发生异常

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

  /*** 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

  private def processLogDirEventNotification(): Unit = {if (!isActive) returnval sequenceNumbers = zkClient.getAllLogDirEventNotificationstry {val brokerIds = zkClient.getBrokerIdsFromLogDirEvents(sequenceNumbers)//尝试将这台Broker上的所有副本 走一下状态流转 到 OnlineReplicaonBrokerLogDirFailure(brokerIds)} finally {// delete processed childrenzkClient.deleteLogDirEventNotifications(sequenceNumbers, controllerContext.epochZkVersion)}}

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

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

源码总结

Q&A