[TOC]
Zookeeper与集群管理原理
我们知道所有的broker启动之后，都会连接到Zookeeper上面。那具体来讲，Zookeeper要帮助Kafka完成什么工作呢？
集群管理的思路
broker的“生“与“死“
任何时候，当集群中有1个新的broker加入，或者某个旧的broker死亡，集群中其它机器都需要知道这件事。
其实现方式就是监听Zookeeper上面的/broker/ids结点，其每个子结点就对应1台broker机器，当broker机器添加，子结点列表增大；broker机器死亡，子结点列表减小。
Controller
为了减小Zookeeper的压力，同时也降低整个分布式系统的复杂度，Kafka引入了一个“中央控制器“，也就是Controller。
其基本思路是：先通过Zookeeper在所有broker中选举出一个Controller，然后用这个Controller来控制其它所有的broker，而不是让zookeeper直接控制所有的机器。
比如上面对/broker/ids的监听，并不是所有broker都监听此结点，而是只有Controller监听此结点，这样就把一个“分布式“问题转化成了“集中式“问题，即降低了Zookeeper负担，也便于控制逻辑的编写。
topic与partition的增加／删除
同样，作为1个分布式集群，当增加／删除一个topic或者partition的时候，不可能挨个通知集群的每1台机器。
这里的实现思路也是：管理端(Admin/TopicCommand)把增加/删除命令发送给Zk，Controller监听Zk获取更新消息, Controller再分发给相关的broker。
I0ITec ZkClient
关于Zookeeper的客户端，我们知道常用的有Apache Curator，但Kafka用的不是这个。而是另外一个叫做I0ITec ZkClient的，记得没错的话，阿里的dubbo框架，也用的这个。相对Curator，它更加轻量级。
具体来说，其主要有3个Listener:
//当某个session断了重连，就会调用这个监听器
public interface IZkStateListener {
public void handleStateChanged(KeeperState state) throws Exception;
public void handleNewSession() throws Exception;
public void handleSessionEstablishmentError(final Throwable error) throws Exception;
}
//当某个结点的data变化之后（data变化，或者结点本事被删除）
public interface IZkDataListener {
public void handleDataChange(String dataPath, Object data) throws Exception;
public void handleDataDeleted(String dataPath) throws Exception;
}
//当某个结点的子结点发生变化
public interface IZkChildListener {
public void handleChildChange(String parentPath, List<String> currentChilds) throws Exception;
}
Kafka正是利用上面3个listener实现了所有zookeeper相关状态变化的监听，其具体应用，将在后续序列逐个展开！
KafkaController选举过程/Failover与Resignation
Kafka集群的几大核心组件
在正式进入源码分析之前，我们先看一下整个Kafka集群的几大核心组件。让我们从整个服务器的main函数开始：
//Kafka
def main(args: Array[String]): Unit = {
try {
val serverProps = getPropsFromArgs(args)
val kafkaServerStartable = KafkaServerStartable.fromProps(serverProps)
// attach shutdown handler to catch control-c
Runtime.getRuntime().addShutdownHook(new Thread() {
override def run() = {
kafkaServerStartable.shutdown //注册一个JVM关闭的钩子
}
})
kafkaServerStartable.startup //启动程序
kafkaServerStartable.awaitShutdown
}
catch {
case e: Throwable =>
fatal(e)
System.exit(1)
}
System.exit(0)
}
//KafkaServerStartable
class KafkaServerStartable(val serverConfig: KafkaConfig) extends Logging {
private val server = new KafkaServer(serverConfig)
def startup() {
try {
server.startup()
}
catch {
case e: Throwable =>
fatal("Fatal error during KafkaServerStartable startup. Prepare to shutdown", e)
// KafkaServer already calls shutdown() internally, so this is purely for logging & the exit code
System.exit(1)
}
}
...
}
//KafkaServer
def startup() {
try {
info("starting")
if(isShuttingDown.get)
throw new IllegalStateException("Kafka server is still shutting down, cannot re-start!")
if(startupComplete.get)
return
val canStartup = isStartingUp.compareAndSet(false, true)
if (canStartup) {
metrics = new Metrics(metricConfig, reporters, kafkaMetricsTime, true)
brokerState.newState(Starting)
//核心组件0
kafkaScheduler.startup()
zkUtils = initZk()
logManager = createLogManager(zkUtils.zkClient, brokerState)
logManager.startup()
config.brokerId = getBrokerId
this.logIdent = "[Kafka Server " + config.brokerId + "], "
//核心组件1
socketServer = new SocketServer(config, metrics, kafkaMetricsTime)
socketServer.startup()
//核心组件2
replicaManager = new ReplicaManager(config, metrics, time, kafkaMetricsTime, zkUtils, kafkaScheduler, logManager,
isShuttingDown)
replicaManager.startup()
//核心组件3
kafkaController = new KafkaController(config, zkUtils, brokerState, kafkaMetricsTime, metrics, threadNamePrefix)
kafkaController.startup()
//核心组件4
consumerCoordinator = GroupCoordinator.create(config, zkUtils, replicaManager)
consumerCoordinator.startup()
/* Get the authorizer and initialize it if one is specified.*/
authorizer = Option(config.authorizerClassName).filter(_.nonEmpty).map { authorizerClassName =>
val authZ = CoreUtils.createObject[Authorizer](authorizerClassName)
authZ.configure(config.originals())
authZ
}
//核心组件5
apis = new KafkaApis(socketServer.requestChannel, replicaManager, consumerCoordinator,
kafkaController, zkUtils, config.brokerId, config, metadataCache, metrics, authorizer)
requestHandlerPool = new KafkaRequestHandlerPool(config.brokerId, socketServer.requestChannel, apis, config.numIoThreads)
brokerState.newState(RunningAsBroker)
Mx4jLoader.maybeLoad()
/* start dynamic config manager */
dynamicConfigHandlers = Map[String, ConfigHandler](ConfigType.Topic -> new TopicConfigHandler(logManager),
ConfigType.Client -> new ClientIdConfigHandler(apis.quotaManagers))
// Apply all existing client configs to the ClientIdConfigHandler to bootstrap the overrides
// TODO: Move this logic to DynamicConfigManager
AdminUtils.fetchAllEntityConfigs(zkUtils, ConfigType.Client).foreach {
case (clientId, properties) => dynamicConfigHandlers(ConfigType.Client).processConfigChanges(clientId, properties)
}
// Create the config manager. start listening to notifications
dynamicConfigManager = new DynamicConfigManager(zkUtils, dynamicConfigHandlers)
dynamicConfigManager.startup()
/* tell everyone we are alive */
val listeners = config.advertisedListeners.map {case(protocol, endpoint) =>
if (endpoint.port == 0)
(protocol, EndPoint(endpoint.host, socketServer.boundPort(protocol), endpoint.protocolType))
else
(protocol, endpoint)
}
kafkaHealthcheck = new KafkaHealthcheck(config.brokerId, listeners, zkUtils)
kafkaHealthcheck.startup()
/* register broker metrics */
registerStats()
shutdownLatch = new CountDownLatch(1)
startupComplete.set(true)
isStartingUp.set(false)
AppInfoParser.registerAppInfo(jmxPrefix, config.brokerId.toString)
info("started")
}
}
catch {
case e: Throwable =>
fatal("Fatal error during KafkaServer startup. Prepare to shutdown", e)
isStartingUp.set(false)
shutdown()
throw e
}
}
除了代码，我们也看一下服务器的启动／关闭的shell脚本：
bin/kafka-server-start.sh
通过 nohup 守护进程，具体脚本细节就不在此列出了。
//bin/kafak-server-stop.sh 可以看到，进程的关闭很简单，就是通过kill命令，发送SIGTERM信号。JVM收到信号，执行上面的钩子函数
ps ax | grep -i 'kafka\.Kafka' | grep java | grep -v grep | awk '{print $1}' | xargs kill -SIGTERM
通过看Server的启动函数，我们可以看到有以下几大核心组件：
1。SocketServer ＋ KafkaApis 前者接收所有网络请求，后者处理请求
2。KafkaController 负责Controller选举
3。ConsumerCoordinator 用于consumer group的负载均衡
4。ReplicaManager 机器的管理
5。KafkaSchedule
这里着重分析KafkaController，其它核心组件，后面会一一讲述。
选举的基本原理
整个选举过程是通过zk上的一个临时节点来实现的：/controller节点，其data结构为：核心信息就是记录当前的controller的brokerId。
"version" -> 1, "brokerid" -> brokerId, "timestamp" -> timestamp
当controller挂了，其它所有broker监听到此临时节点消失，然后争相创建此临时节点，谁创建成功，谁就成为新的Controller。
除了/controller节点，还有一个辅助的/controller_epoch，记录当前Controller的轮值数。
KafkaController与ZookeeperLeaderElector
整个选举过程是通过这2个核心类实现的，其中ZookeeperLeaderElector是KafkaController的一个成员变量：
//KafkaController的一个成员变量
private val controllerElector = new ZookeeperLeaderElector(controllerContext, ZkUtils.ControllerPath, onControllerFailover,
onControllerResignation, config.brokerId)
下图展示了选举的整个交互过程：
（1）KafkaController和ZookeeperLeaderElector内部各有1个Listener，一个监听session重连，1个监听/controller节点变化。
（2）当session重连，或者/controller节点被删除，则调用elect()函数，发起重新选举。在重新选举之前，先判断自己是否旧的Controller，如果是，则先调用onResignation退位。
image.png
下面从KakfaController的startup函数看起：
def startup() = {
inLock(controllerContext.controllerLock) {
info("Controller starting up")
registerSessionExpirationListener() //第1种监听：SessionExpirationListener
isRunning = true
controllerElector.startup //第2种监听：LeaderChangeListener
info("Controller startup complete")
}
}
class SessionExpirationListener() extends IZkStateListener with Logging {
...
@throws(classOf[Exception])
def handleNewSession() {
info("ZK expired; shut down all controller components and try to re-elect")
inLock(controllerContext.controllerLock) {
onControllerResignation() //先退位
controllerElector.elect //发起重新选举
}
}
...
}
class LeaderChangeListener extends IZkDataListener with Logging {
@throws(classOf[Exception])
def handleDataChange(dataPath: String, data: Object) {
inLock(controllerContext.controllerLock) {
val amILeaderBeforeDataChange = amILeader
leaderId = KafkaController.parseControllerId(data.toString)
if (amILeaderBeforeDataChange && !amILeader)
onResigningAsLeader() //自己以前是controller，现在不是，退位
}
}
@throws(classOf[Exception])
def handleDataDeleted(dataPath: String) {
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() //关键点：controller死了，有可能不是因为自己死了。而是和zookeeper的session断了。但是自己还在。此时，自己先退休，再重新发起选举。
elect //发起重现选举
}
}
}
2个关键回调：Failover(上任)与Resignation(退位)
在上面的选举过程中，存在2个关键的callback：也就是新Controller上任要做的事情和旧Controller退位要做的事情。
“上任“这个比较容易理解，也就是新的broker选举为controller；那为什么会有“退位“呢？
这是因为zk是用心跳来判断controller是否存活，可能controller存活，但zk认为它挂了，这个时候选举出了新的controller。那旧的controller发现自己是旧的，就得主动退位。
下面看一下“新官上任“和“旧官退位“时，分别做了什么：
def onControllerFailover() {
if(isRunning) {
readControllerEpochFromZookeeper()
//递增controller epoch
incrementControllerEpoch(zkUtils.zkClient)
//关键点：接管所有对broker/partition节点的监听
registerReassignedPartitionsListener()
registerIsrChangeNotificationListener()
registerPreferredReplicaElectionListener()
partitionStateMachine.registerListeners()
replicaStateMachine.registerListeners()
initializeControllerContext()
replicaStateMachine.startup()
partitionStateMachine.startup()
// register the partition change listeners for all existing topics on failover
controllerContext.allTopics.foreach(topic => partitionStateMachine.registerPartitionChangeListener(topic))
info("Broker %d is ready to serve as the new controller with epoch %d".format(config.brokerId, epoch))
brokerState.newState(RunningAsController)
maybeTriggerPartitionReassignment()
maybeTriggerPreferredReplicaElection()
/* send partition leadership info to all live brokers */
sendUpdateMetadataRequest(controllerContext.liveOrShuttingDownBrokerIds.toSeq)
if (config.autoLeaderRebalanceEnable) {
info("starting the partition rebalance scheduler")
autoRebalanceScheduler.startup()
autoRebalanceScheduler.schedule("partition-rebalance-thread", checkAndTriggerPartitionRebalance,
5, config.leaderImbalanceCheckIntervalSeconds.toLong, TimeUnit.SECONDS)
}
deleteTopicManager.start()
}
else
info("Controller has been shut down, aborting startup/failover")
}
def onControllerResignation() {
//关键点：放弃对所有broker/partition的监听
deregisterIsrChangeNotificationListener()
deregisterReassignedPartitionsListener()
deregisterPreferredReplicaElectionListener()
// shutdown delete topic manager
if (deleteTopicManager != null)
deleteTopicManager.shutdown()
// shutdown leader rebalance scheduler
if (config.autoLeaderRebalanceEnable)
autoRebalanceScheduler.shutdown()
inLock(controllerContext.controllerLock) {
// de-register partition ISR listener for on-going partition reassignment task
deregisterReassignedPartitionsIsrChangeListeners()
// shutdown partition state machine
partitionStateMachine.shutdown()
// shutdown replica state machine
replicaStateMachine.shutdown()
// shutdown controller channel manager
if(controllerContext.controllerChannelManager != null) {
controllerContext.controllerChannelManager.shutdown()
controllerContext.controllerChannelManager = null
}
// reset controller context
controllerContext.epoch=0
controllerContext.epochZkVersion=0
brokerState.newState(RunningAsBroker)
}
}
KafkaController成为leader的过程
KafkaController内部有一个ZookeeperLeaderElector，用来通过zk选举自己是否是leader
class KafkaController(val config : KafkaConfig, zkClient: ZkClient, val brokerState: BrokerState) extends Logging with KafkaMetricsGroup {
……
private val controllerElector = new ZookeeperLeaderElector(controllerContext, ZkUtils.ControllerPath, onControllerFailover,
onControllerResignation, config.brokerId)
/**
* Invoked when the controller module of a Kafka server is started up. This does not assume that the current broker
* is the controller. It merely registers the session expiration listener and starts the controller leader
* elector
*/
def startup() = {
inLock(controllerContext.controllerLock) {
info("Controller starting up");
registerSessionExpirationListener()//注册一个会话超时的listener
isRunning = true
controllerElector.startup//启动controllerElector
info("Controller startup complete")
}
}
}
其zk选举的路径为/controller/*，并且对zk集群建立一个会话超时的listener
class SessionExpirationListener() extends IZkStateListener with Logging {
this.logIdent = "[SessionExpirationListener on " + config.brokerId + "], "
@throws(classOf[Exception])
def handleStateChanged(state: KeeperState) {
// do nothing, since zkclient will do reconnect for us.
}
/**
* Called after the zookeeper session has expired and a new session has been created. You would have to re-create
* any ephemeral nodes here.
*
* @throws Exception
* On any error.
*/
@throws(classOf[Exception])
def handleNewSession() {
info("ZK expired; shut down all controller components and try to re-elect")
inLock(controllerContext.controllerLock) {
onControllerResignation()//当会话超时，重新连接上的时候，调用之前注册在ZookeeperLeaderElector的onControllerResignation函数
controllerElector.elect//重新选举
}
}
}
因此重点关注ZookeeperLeaderElector内部的逻辑：
class ZookeeperLeaderElector(controllerContext: ControllerContext,
electionPath: String,
onBecomingLeader: () => Unit,
onResigningAsLeader: () => Unit,
brokerId: Int)
extends LeaderElector with Logging {
var leaderId = -1
// create the election path in ZK, if one does not exist
val index = electionPath.lastIndexOf("/")
if (index > 0)
makeSurePersistentPathExists(controllerContext.zkClient, electionPath.substring(0, index))
val leaderChangeListener = new LeaderChangeListener
def startup {
inLock(controllerContext.controllerLock) {//其选举路径为/controller/*
controllerContext.zkClient.subscribeDataChanges(electionPath, leaderChangeListener)
elect//触发选举
}
}
private def getControllerID(): Int = {
readDataMaybeNull(controllerContext.zkClient, electionPath)._1 match {
case Some(controller) => KafkaController.parseControllerId(controller)
case None => -1
}
}
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 {//通过zk创建Ephemeral Node的方式来进行选举，即如果存在并发情况下向zk的同一个路径创建node的话，有且只有1个客户端会创建成功，其它客户端创建失败，但是当创建成功的客户端和zk的链接断开之后，这个node也会消失，其它的客户端从而继续竞争
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()//如果成功，则自己成为leader
} 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
}
def close = {
leaderId = -1
}
def amILeader : Boolean = leaderId == brokerId
def resign() = {
leaderId = -1
deletePath(controllerContext.zkClient, electionPath)
}
/**
* We do not have session expiration listen in the ZkElection, but assuming the caller who uses this module will
* have its own session expiration listener and handler
*/
class LeaderChangeListener extends IZkDataListener with Logging {
/**
* Called when the leader information stored in zookeeper has changed. Record the new leader in memory
* @throws Exception On any error.
*/
@throws(classOf[Exception])
def handleDataChange(dataPath: String, data: Object) {
inLock(controllerContext.controllerLock) {
leaderId = KafkaController.parseControllerId(data.toString)
info("New leader is %d".format(leaderId))
}
}
/**
* Called when the leader information stored in zookeeper has been delete. Try to elect as the leader
* @throws Exception
* On any error.
*/
@throws(classOf[Exception])
def handleDataDeleted(dataPath: String) {//KafkaController在第一次启动的时候没有选举成功，然后当其发现节点已经消失的时候，会重新触发选举
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)//可能之前自己的角色是leader，则重新选举未必成为leader，则需要清除之前所有缓存的内容
onResigningAsLeader()
elect//触发选举
}
}
}
}
因此KafkaController成为leader分2种情况：
- 第一次启动的时候会主动触发elect，如果被选举成为leader，则做leader该做的事情
- 第一次启动的时候选举失败，则通过LeaderChangeListener监控/controller/*路径，发现下面数据被删除的时候，触发handleDataDeleted，从而再次触发选举
12.2 kafkaController的初始化（leader）
从上节可以看到，KafkaController选举成功则调用onBecomingLeader，当之前的leader再次触发选举的时候调用onResigningAsLeader，以上2个函数分别对应：onControllerFailover和onControllerResignation。
onControllerResignation很简单，就是把里面所有的模块shutdown或者注销掉：
def onControllerResignation() {
// de-register listeners
deregisterReassignedPartitionsListener()
deregisterPreferredReplicaElectionListener()
// shutdown delete topic manager
if (deleteTopicManager != null)
deleteTopicManager.shutdown()
// shutdown leader rebalance scheduler
if (config.autoLeaderRebalanceEnable)
autoRebalanceScheduler.shutdown()
inLock(controllerContext.controllerLock) {
// de-register partition ISR listener for on-going partition reassignment task
deregisterReassignedPartitionsIsrChangeListeners()
// shutdown partition state machine
partitionStateMachine.shutdown()
// shutdown replica state machine
replicaStateMachine.shutdown()
// shutdown controller channel manager
if(controllerContext.controllerChannelManager != null) {
controllerContext.controllerChannelManager.shutdown()
controllerContext.controllerChannelManager = null
}
// reset controller context
controllerContext.epoch=0
controllerContext.epochZkVersion=0
brokerState.newState(RunningAsBroker)
}
}
以上各种模块会在onControllerFailover介绍，onControllerFailover本质上就是开启里面所有的功能。
onControllerFailover的逻辑如下：
def onControllerFailover() {
if(isRunning) {
info("Broker %d starting become controller state transition".format(config.brokerId))
readControllerEpochFromZookeeper()
//记录选举的时钟，每成功选举一次，递增1
incrementControllerEpoch(zkClient)
/*leader初始化，具体内容见评注*/
registerReassignedPartitionsListener()
registerPreferredReplicaElectionListener()
partitionStateMachine.registerListeners()
replicaStateMachine.registerListeners()
initializeControllerContext()
replicaStateMachine.startup()
partitionStateMachine.startup()
controllerContext.allTopics.foreach(topic => partitionStateMachine.registerPartitionChangeListener(topic))
info("Broker %d is ready to serve as the new controller with epoch %d".format(config.brokerId, epoch))
brokerState.newState(RunningAsController)
maybeTriggerPartitionReassignment()
maybeTriggerPreferredReplicaElection()
sendUpdateMetadataRequest(controllerContext.liveOrShuttingDownBrokerIds.toSeq)
if (config.autoLeaderRebalanceEnable) {
info("starting the partition rebalance scheduler")
autoRebalanceScheduler.startup()
autoRebalanceScheduler.schedule("partition-rebalance-thread", checkAndTriggerPartitionRebalance,
5, config.leaderImbalanceCheckIntervalSeconds, TimeUnit.SECONDS)
}
deleteTopicManager.start()
}
else
info("Controller has been shut down, aborting startup/failover")
}
其中步骤如下：
1） 在/admin/reassign_partitions目录注册partitionReassignedListener监听函数
2） 在/admin/preferred_replica_election目录注册preferredReplicaElectionListener监听函数
3） 在/brokers/topics目录注册topicChangeListener监听函数
4） 在/admin/delete_topics目录注册deleteTopicsListener监听函数
5） 在/brokers/ids目录注册brokerChangeListener监听函数
6） 初始化ControllerContext上下文，里面包含了topic的各种元数据信息，除此之外ControllerContext内部的ControllerChannelManager负责和kafka集群内部的其它KafkaServer建立channel来进行通信，TopicDeletionManager
负责删除topic
7）通过replicaStateMachine初始化所有的replica状态
8）通过partitionStateMachine初始化所有的partition状态
- 在brokers/topics/***(具体的topic名字)/目录下注册AddPartitionsListener函数
- 通过处理之前启动留下的partition重分配的情况
- 处理之前启动留下的replica重新选举的情况
12）向其它KafkaServer发送集群topic的元数据信息已进行数据的同步更新
13）根据配置是否开启自动均衡
14）开始删除topic
KafkaControl主要通过以上各种监听函数来完成kafka集群元数据的管理，接下来先详细描述PartitionStateMachine和ReplicaStateMachine原理，因为kafka topic 的partition状态和内容主要是通过以上2个管理类来实现的，然后按照上面的流程描述不同的listener的作用。
网友评论