PartitionStateMachine是Controller Leader用于维护分区状态的状态机。分区的状态是通过PartitionState接口定义的，它有四个子类分别代表了分区四种可能的状态，如表所示。
分区各个PartitionState之间的转换如图所示。
下面分析各个状态之间转换时，需要完成的相关操作。
- NonExistentPartition →NewPartition
从ZooKeeper中加载分区的AR集合到ControllerContext的partitionReplicaAssignment集合中。
- NewPartition →OnlinePartition
首先将Leader副本和ISR集合的信息写入到ZooKeeper中，这里会将分区的AR集合中第一个可用的副本选举为Leader副本，并将分区的所有可用副本作为ISR集合。
之后，向所有可用的副本发送LeaderAndIsrRequest,指导这些副本进行Leader/Follower的角色切换，并向所有可用的Broker发送UpdateMetadataRequest来更新其上的MetadataCache。
- OnlinePartitio/OffinePartition →OnlinePartition
为分区选择新的Leader副本和ISR集合，并将结果写入ZooKeeper。之后，向需要进行角色切换的副本发送LeaderAndIsrRequest,指导这些副本进行Leader/Follower的角色切换，并向所有可用的Broker发送UpdateMetadataRequest来更新其上的MetadataCache。
- NewPartition,OnlinePartition →OfflinePartition
只进行状态转换，并没有其他的操作。
- OfinePartition →NonExistentPartition
只进行状态转换，并没有其他的操作。
PartitionStateMachine中的各个字段含义和作用如下所述。
- controllerContext:ControllerContext对象，用于维护KafkaController的上下文信息。
- zkUtils:ZooKeeper的客户端，用于与ZooKeeper服务器交互。
- partitionState:Map[TopicAndPartition,PartitionState]类型，记录了每个分区对应的PartitionState状态。
- brokerRequestBatch:ControllerBrokerRequestBatch对象，用于向指定的Broker批量发送请求。
noOpPartitionLeaderSelector: 默认的Leader副本选举类器， 继承了PartitionLeaderSelector。NoOpLeaderSelector实现并没有真正进行Leader副本的选举，其实现是返回当前的Leader副本、ISR集合和AR集合。
- topicChangeListener:ZooKeeper的监听器，用于监听Topic的变化。
- deleteTopicsListener:ZooKeeper的监听器，用于监听Topic的删除。
- partitionModificationsListeners:用于监听分区的修改。
PartitionStateMachine启动时会对partitionState集合进行初始化，并调用triggerOnlinePartitionStateChange方法将NewPartition和OfflinePartition状态的分区转换成OnlinePartition状态。
PartitionStateMachine.handleStateChange()方法是管理分区状态的核心方法，该方法控制着PartitionState的转换。这里需要注意该方法的第三个参数，它指定了用来选举Leader副本的PartitionLeaderSelector对象。
PartitionState由NewPartition切换为OnlinePartition时，调用了initializeLeaderAndIsrForPartition方法，其操作的主要步骤是：
- 从ControllerContext.partitionReplicaAssignment集合中选择第一个可用的副本作为Leader副本，其余的副本构成ISR集合。
- 将Leader副本和ISR集合的信息写入到ZooKeeper。
- 更新ControllerContext.partitionLeadershipInfo中缓存的Leader副本、ISR集合等信息。
- 将上述步骤中确定的Leader副本、ISR集合、AR集合等信息添加到ControllerBrokerRequestBatch,之后会封装成LeaderAndlsrRequest发送给相关的Broker。
