Kafka 源码解析：Broker 节点的启动与关闭

def main(args: Array[String]): Unit = {
    try {
        // 解析命令行参数
        val serverProps = getPropsFromArgs(args)
        // 创建 kafkaServerStartable 对象，期间会初始化监控上报程序
        val kafkaServerStartable = KafkaServerStartable.fromProps(serverProps)

        // 注册一个钩子方法，当 JVM 被关闭时执行 shutdown 逻辑，本质上是在执行 KafkaServer#shutdown 方法
        Runtime.getRuntime.addShutdownHook(new Thread() {
            override def run(): Unit = {
                kafkaServerStartable.shutdown()
            }
        })

        // 本质上调用的是 KafkaServer#startup 方法
        kafkaServerStartable.startup()
        // 阻塞等待 kafka server 运行线程关闭
        kafkaServerStartable.awaitShutdown()
    } catch {
        case e: Throwable =>
            fatal(e)
            System.exit(1)
    }
    System.exit(0)
}

class KafkaServer(val config: KafkaConfig, // 配置信息对象
                  time: Time = Time.SYSTEM, // 时间戳工具
                  threadNamePrefix: Option[String] = None,
                  kafkaMetricsReporters: Seq[KafkaMetricsReporter] = List() // 监控上报程序
                 ) extends Logging with KafkaMetricsGroup {

    /** 标识节点已经启动完成 */
    private val startupComplete = new AtomicBoolean(false)
    /** 标识节点正在执行关闭操作 */
    private val isShuttingDown = new AtomicBoolean(false)
    /** 标识节点正在执行启动操作 */
    private val isStartingUp = new AtomicBoolean(false)
    /** 阻塞主线程等待 KafkaServer 的关闭 */
    private var shutdownLatch = new CountDownLatch(1)
    /** 记录 broker 节点的当前状态 */
    val brokerState: BrokerState = new BrokerState
    /** Api 接口类，用于分发各种类型的请求 */
    var apis: KafkaApis = _
    /** 权限控制相关 */
    var authorizer: Option[Authorizer] = None
    var credentialProvider: CredentialProvider = _
    /** 网络 socket 服务 */
    var socketServer: SocketServer = _
    /** 简单的连接池实现，用于管理所有的 KafkaRequestHandler */
    var requestHandlerPool: KafkaRequestHandlerPool = _
    /** 日志数据管理 */
    var logManager: LogManager = _
    /** 管理当前 broker 节点上的分区副本 */
    var replicaManager: ReplicaManager = _
    /** topic 增删管理 */
    var adminManager: AdminManager = _
    /** 动态配置管理 */
    var dynamicConfigHandlers: Map[String, ConfigHandler] = _
    var dynamicConfigManager: DynamicConfigManager = _
    /** group 协调管理组件 */
    var groupCoordinator: GroupCoordinator = _
    /** 集群控制组件 */
    var kafkaController: KafkaController = _
    /** 定时任务调度器 */
    val kafkaScheduler = new KafkaScheduler(config.backgroundThreads)
    /** broker 节点活跃性检查 */
    var kafkaHealthcheck: KafkaHealthcheck = _
    /** broker 缓存整个集群中全部分区的状态信息 */
    var metadataCache: MetadataCache = _
    /** ZK 操作 工具 类 */
    var zkUtils: ZkUtils = _

    // ... 省略方法定义

}

sealed trait BrokerStates { def state: Byte }

case object NotRunning extends BrokerStates { val state: Byte = 0 }
case object Starting extends BrokerStates { val state: Byte = 1 }
case object RecoveringFromUncleanShutdown extends BrokerStates { val state: Byte = 2 }
case object RunningAsBroker extends BrokerStates { val state: Byte = 3 }
case object PendingControlledShutdown extends BrokerStates { val state: Byte = 6 }
case object BrokerShuttingDown extends BrokerStates { val state: Byte = 7 }

trait Scheduler {
    def startup()
    def shutdown()
    def isStarted: Boolean
    def schedule(name: String, fun: () => Unit, delay: Long = 0, period: Long = -1, unit: TimeUnit = TimeUnit.MILLISECONDS)
}

private def getBrokerId: Int = {
    // 获取配置的 brokerId
    var brokerId = config.brokerId
    val brokerIdSet = mutable.HashSet[Int]()

    // 遍历 log.dirs 配置的 log 目录列表
    for (logDir <- config.logDirs) {
        // 在每一个 log 目录下面创建一个 meta.properties 文件，内容包含当前 broker 节点的 ID 和版本信息
        val brokerMetadataOpt = brokerMetadataCheckpoints(logDir).read()
        brokerMetadataOpt.foreach { brokerMetadata =>
            brokerIdSet.add(brokerMetadata.brokerId)
        }
    }

    if (brokerIdSet.size > 1) {
        // 不允许多个 broker 节点共享同一个 log 目录
        // ... 抛出 InconsistentBrokerIdException 异常，略
    } else if (brokerId >= 0 && brokerIdSet.size == 1 && brokerIdSet.last != brokerId) {
        // 配置的 brokerId 与 meta.properties 中记录的 brokerId 不一致
        // ... 抛出 InconsistentBrokerIdException 异常，略
    } else if (brokerIdSet.isEmpty && brokerId < 0 && config.brokerIdGenerationEnable) {
        // 如果没有配置，则自动创建 brokerId，通过 ZK 保证 brokerId 的全局唯一性
        brokerId = generateBrokerId
    } else if (brokerIdSet.size == 1) {
        // 从 meta.properties 中获取 brokerId
        brokerId = brokerIdSet.last
    }

    brokerId
}

def shutdown() {
    try {
        info("shutting down")

        // 如果正在启动，则不允许关闭
        if (isStartingUp.get)
            throw new IllegalStateException("Kafka server is still starting up, cannot shut down!")

        if (shutdownLatch.getCount > 0 && isShuttingDown.compareAndSet(false, true)) {
            CoreUtils.swallow(controlledShutdown())
            // 设置 broker 状态为 BrokerShuttingDown，表示当前 broker 正在执行关闭操作
            brokerState.newState(BrokerShuttingDown)

            /* 依次关闭相应注册的组件 */

            if (socketServer != null) CoreUtils.swallow(socketServer.shutdown())
            if (requestHandlerPool != null) CoreUtils.swallow(requestHandlerPool.shutdown())
            CoreUtils.swallow(kafkaScheduler.shutdown())
            if (apis != null) CoreUtils.swallow(apis.close())
            CoreUtils.swallow(authorizer.foreach(_.close()))
            if (replicaManager != null) CoreUtils.swallow(replicaManager.shutdown())
            if (adminManager != null) CoreUtils.swallow(adminManager.shutdown())
            if (groupCoordinator != null) CoreUtils.swallow(groupCoordinator.shutdown())
            if (logManager != null) CoreUtils.swallow(logManager.shutdown())
            if (kafkaController != null) CoreUtils.swallow(kafkaController.shutdown())
            if (zkUtils != null) CoreUtils.swallow(zkUtils.close())
            if (metrics != null) CoreUtils.swallow(metrics.close())

            // 设置 broker 状态为 NotRunning，表示关闭成功
            brokerState.newState(NotRunning)

            // 设置状态标记
            startupComplete.set(false)
            isShuttingDown.set(false)
            CoreUtils.swallow(AppInfoParser.unregisterAppInfo(jmxPrefix, config.brokerId.toString))
            shutdownLatch.countDown()
            info("shut down completed")
        }
    } catch {
        case e: Throwable =>
            fatal("Fatal error during KafkaServer shutdown.", e)
            isShuttingDown.set(false)
            throw e
    }
}