是
ShutdownableThread
Zookeeper Cluster
Register local operation
return
写入事件
handleCreationhandleDataChangehandleDeletion
invoke
maybeCreateControllerEpochZNode持久化节点 /controller_epoch
阻塞队列 queue
doWork()
elect()
take
KafkaController
ZooKeeperClientWatcher
startup
dequeued.process(processor)
事件监听
ControllerEventManager
否
ControllerEventManagerThread start
ControllerEventManagerput Start Event
extend
activeControllerId = zkClient.getControllerId
ControllerEventThread
run()
start
activeControllerId不等于 1
KafkaController 实现了 ControllerEventProcessor
tryCreateControllerZNodeAndIncrementEpoch临时节点 /controller
ExistsRequest/GetDataRequestwatch 节点
ZookeeperClient.handleRequests
processStartup()
Zookeeper client
KafkaServer
KafkaZkClient
ZookeeperClient
