15 | 如何理解Controller在Kafka集群中的作用？

你好，我是胡夕。我来公布上节课的“课后讨论”题答案啦～ 上节课，咱们重点了解了Controller选举部分的代码。课后我请你思考这样一个问题：源码中当Controller选举之后哪里更新的元数据请求？ 其实这是在onControllerFailover方法中完成的。该方法中调用： sendUpdateMetadataRequest(controllerContext.liveOrShuttingDownBrokerIds.toSeq, Set.empty)令集群所有Broker更新元数据 okay，你同意这个说法吗？或者说你有其他的看法吗？我们可以一起讨论下。

老师，学习了关于Controller源码课，我再深入看相关源码，梳理了流程图，您看这流程图是否正确 http://note.youdao.com/noteshare?id=1f999786b77ada75fbb58b2831770e47&sub=554F637E4F984C7B8036365C51E0758B

老师，在看 Controller 处理各种事件的时候，发现相关的处理方法中都没有重新注册watcher。比如新启动节点触发了 BrokerChange 事件，处理流程中并没有重新注册对/brokers/ids 子节点的监听。这个是在哪里进行重新注册的

胡夕老师，我们项目最近在预研使用图数据库，刚看碰巧在一篇文章（https://zhuanlan.zhihu.com/p/99381529） 看到你们公司使用JanusGraph，想问下这数据库稳定性如何，同时也看到另外一款百度开源的HugeGraph， 你们选型的JanusGraph看中的优势是什么？

调用脚本kafka-topics.sh exec $(dirname $0)/kafka-run-class.sh kafka.admin.TopicCommand "$@" 对应的类object TopicCommand 的topicService.createTopic(opts)根据是否定义ZK信息走实现类ZookeeperTopicService或者AdminClientTopicService然后 ZookeeperTopicService里面 adminZkClient.createTopic 然后就可以触发topicChange事件...

为什么新增的broker马上就分配好了副本呢？什么时候分配的？

这节课应该往前提，提到这个主题的前面，讲的太好了

胡夕老师，最近遇到个问题，向请教一下您： 我下线了controller节点1后，2被选为新的controller。但是发现有个topic A的partition1元数据信息不对，原先它的三个副本在2，3，4（2为leader）。但是看打印的日志中发现新controller 2 告诉6节点说topic A的partition1的副本是3，2，6，但并没有向2，3，4同步元数据的变更，导致6节点无故新增了一副本并向leader拉取一直失败，以下是日志：您帮我看看这是为什么呢？感觉像是controller的下线导致元数据错乱了 6机器上的日志： [2021-03-16 14:12:39.979] TRACE [Broker id=6] Cached leader info UpdateMetadataPartitionState(topicName='A', p artitionIndex=1, controllerEpoch=29, leader=2, leaderEpoch=13, isr=[2, 3, 4], zkVersion=25, replicas=[2, 3, 4], offlineReplicas=[ ]) for partition A-1 in response to UpdateMetadata request sent by controller 1 epoch 29 with correlation id 0 (state.change.logger) [2021-03-16 15:00:46.600] TRACE [Broker id=6] Cached leader info UpdateMetadataPartitionState(topicName='A', p artitionIndex=1, controllerEpoch=29, leader=2, leaderEpoch=13, isr=[2, 3, 4], zkVersion=25, replicas=[3, 2, 6], offlineReplicas=[ ]) for partition A-1 in response to UpdateMetadata request sent by controller 2 epoch 30 with correlation id 0 (state.change.logger) 2机器上的日志： [2021-03-16 15:07:26,164] WARN [ReplicaManager broker=2] Leader 2 failed to record follower 6's position 11679162 since the replica is not recognized to be one of the assigned replicas 3,4,2 for partition A-1. Empty records will be returned for this partition. (kafka.server.ReplicaManager)

增加一个主题的分区调用的是 KafkaController 的方法: processTopicChange -> onNewPartitionCreation

调用的是onNewPartitionCreation方法，源码如下 private def onNewPartitionCreation(newPartitions: Set[TopicPartition]): Unit = { info(s"New partition creation callback for ${newPartitions.mkString(",")}") partitionStateMachine.handleStateChanges(newPartitions.toSeq, NewPartition) replicaStateMachine.handleStateChanges(controllerContext.replicasForPartition(newPartitions).toSeq, NewReplica) partitionStateMachine.handleStateChanges( newPartitions.toSeq, OnlinePartition, Some(OfflinePartitionLeaderElectionStrategy(false)) ) replicaStateMachine.handleStateChanges(controllerContext.replicasForPartition(newPartitions).toSeq, OnlineReplica) }