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

我会带着你远行 2022-10-09 03:19 52阅读 0赞

**前言**

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

## 源码分析 ##

这里说的dirLog是 server.properties中配置的`log.dir` 例如  
![在这里插入图片描述][20210623175508630.png]

### 副本异常处理 ###

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

`ReplicaManager.startup()`

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)
        }
      }

// 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`
    
        {"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) 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 ##

[20210623175508630.png]: /images/20221005/c85adbd8cd1647a58dcbd58c6210eaee.png