PartitionStateMachine

PartitionStateMachine is the base of Partition State Machines that can handleStateChanges, be started up and shut down in the end.

Table 1. PartitionStateMachine Contract (Abstract Methods Only)
Method	Description
`handleStateChanges`	`handleStateChanges( partitions: Seq[TopicPartition], targetState: PartitionState, leaderElectionStrategy: Option[PartitionLeaderElectionStrategy] ): Map[TopicPartition, Throwable]` Handles state changes of partitions (partition state changes) PartitionLeaderElectionStrategy is undefined (`None`) by default Used when: `KafkaController` is requested to onReplicasBecomeOffline, onNewPartitionCreation, onNewPartitionCreation, onPreferredReplicaElection, moveReassignedPartitionLeaderIfRequired, and doControlledShutdown `PartitionStateMachine` is requested to triggerOnlineStateChangeForPartitions `TopicDeletionManager` is requested to onTopicDeletion

Note	ZkPartitionStateMachine is the default and only known implementation of the PartitionStateMachine Contract in Apache Kafka.

PartitionStateMachine takes a single ControllerContext to be created.

Note	`PartitionStateMachine` is a Scala abstract class and cannot be created directly. It is created indirectly for the concrete PartitionStateMachines.

PartitionLeaderElectionStrategy

Table 2. PartitionLeaderElectionStrategies
Name	Description
`ControlledShutdownPartitionLeaderElectionStrategy`
`OfflinePartitionLeaderElectionStrategy`	Accepts `allowUnclean` flag Handled by `ZkPartitionStateMachine` when requested to doElectLeaderForPartitions Used when: `KafkaController` is requested to onNewPartitionCreation (with the `allowUnclean` flag off), onReplicaElection (with the `allowUnclean` flag on for the admin client) `PartitionStateMachine` is requested to triggerOnlineStateChangeForPartitions (with the `allowUnclean` flag off)
`PreferredReplicaPartitionLeaderElectionStrategy`	`KafkaController` is requested for preferred replica leader election that in turn triggers `ZkPartitionStateMachine` to leaderForPreferredReplica
`ReassignPartitionLeaderElectionStrategy`

`triggerOnlinePartitionStateChange` Method

triggerOnlinePartitionStateChange(): Unit
triggerOnlinePartitionStateChange(
  topic: String): Unit (1)

Uses the partitions of the given topic only

triggerOnlinePartitionStateChange requests the ControllerContext for the partitions in the states: OfflinePartition and NewPartition.

With the topic specified, triggerOnlinePartitionStateChange requests the ControllerContext for the partitions of the topic in the states: OfflinePartition and NewPartition.

In the end, triggerOnlinePartitionStateChange triggers online state change for the partitions.

Note	`triggerOnlinePartitionStateChange` is used when: `KafkaController` is requested to onBrokerStartup, onReplicasBecomeOffline, processUncleanLeaderElectionEnable, processTopicUncleanLeaderElectionEnable `PartitionStateMachine` is requested to start up

`triggerOnlineStateChangeForPartitions` Internal Method

triggerOnlineStateChangeForPartitions(
  partitions: collection.Set[TopicPartition]): Unit

triggerOnlineStateChangeForPartitions filters out the partitions of the topics to be deleted and tries to move the partitions to OnlinePartition state with OfflinePartitionLeaderElectionStrategy (with allowUnclean flag off).

Note	`triggerOnlineStateChangeForPartitions` is used when `PartitionStateMachine` is requested to triggerOnlinePartitionStateChange.

Shutting Down — `shutdown` Method

shutdown(): Unit

shutdown simply prints out the following INFO message to the logs:

Stopped partition state machine

Note	`shutdown` is used exclusively when is requested to resign as the active controller

Starting Up (On Active Controller) — `startup` Method

startup(): Unit

startup prints out the following INFO message to the logs:

Initializing partition state

startup initializePartitionState.

startup prints out the following INFO message to the logs:

Triggering online partition state changes

startup triggerOnlinePartitionStateChange.

In the end, startup prints out the following INFO message to the logs:

Started partition state machine with initial state -> [partitionStates]

Note	`startup` is used exclusively when `KafkaController` is requested to onControllerFailover (when a broker is successfully elected as the controller).

`initializePartitionState` Internal Method

initializePartitionState(): Unit

initializePartitionState requests the ControllerContext for all partitions (across all the brokers in the Kafka cluster).

For every TopicPartition, initializePartitionState requests the ControllerContext for the LeaderIsrAndControllerEpoch metadata (using the partitionLeadershipInfo internal registry).

initializePartitionState changeStateTo of a TopicPartition as follows:

OnlinePartition when the ControllerContext says that the replica is online (for the leader ISR and the TopicPartition)
OfflinePartition when the ControllerContext says that the replica is not online (for the leader ISR and the TopicPartition)
NewPartition when the ControllerContext has no metadata about the TopicPartition

Note	`initializePartitionState` is used exclusively when `PartitionStateMachine` is requested to start up on the active controller.

PartitionStateMachine