ControllerContext

ControllerContext is the context of an active KafkaController (and is created right when KafkaController is created).

ControllerContext takes no input arguments to be created.

`allPartitions` Method

allPartitions: Set[TopicPartition]

allPartitions converts the partitionReplicaAssignmentUnderlying into TopicPartitions, i.e. allPartitions takes the partitions for the topics and simply creates new TopicPartitions.

Note	`allPartitions` is used when: `KafkaController` is requested to updateLeaderAndIsrCache, checkAndTriggerAutoLeaderRebalance, and updateMetrics `PartitionStateMachine` is requested to initializePartitionState `ReplicaStateMachine` is requested to initializeReplicaState

`updatePartitionReplicaAssignment` Method

updatePartitionReplicaAssignment(topicPartition: TopicPartition, newReplicas: Seq[Int]): Unit

updatePartitionReplicaAssignment simply updates the partitionReplicaAssignmentUnderlying registry with newReplicas for the topic and the partition (of a given TopicPartition).

Note

updatePartitionReplicaAssignment is used when:

KafkaController is requested to initializeControllerContext, moveReassignedPartitionLeaderIfRequired, updateAssignedReplicasForPartition, and at TopicChange and PartitionModifications controller events
ReplicaStateMachine is requested to doHandleStateChanges

`partitionsOnBroker` Method

partitionsOnBroker(brokerId: Int): Set[TopicPartition]

partitionsOnBroker…FIXME

Note	`partitionsOnBroker` is used when…FIXME

`replicasOnBrokers` Method

replicasOnBrokers(
  brokerIds: Set[Int]): Set[PartitionAndReplica]

replicasOnBrokers…FIXME

Note	`replicasOnBrokers` is used when…FIXME

`replicasForTopic` Method

replicasForTopic(topic: String): Set[PartitionAndReplica]

replicasForTopic…FIXME

Note	`replicasForTopic` is used when…FIXME

`partitionsForTopic` Method

partitionsForTopic(topic: String): collection.Set[TopicPartition]

partitionsForTopic…FIXME

Note	`partitionsForTopic` is used when…FIXME

`removeTopic` Method

removeTopic(topic: String): Unit

removeTopic…FIXME

Note	`removeTopic` is used when…FIXME

`isReplicaOnline` Method

isReplicaOnline(
  brokerId: Int,
  topicPartition: TopicPartition,
  includeShuttingDownBrokers: Boolean = false): Boolean

isReplicaOnline…FIXME

Note	`isReplicaOnline` is used when…FIXME

`allLiveReplicas` Method

allLiveReplicas(): Set[PartitionAndReplica]

allLiveReplicas…FIXME

Note	`allLiveReplicas` is used when…FIXME

`onlineAndOfflineReplicas` Method

onlineAndOfflineReplicas: (Set[PartitionAndReplica], Set[PartitionAndReplica])

onlineAndOfflineReplicas…FIXME

Note	`onlineAndOfflineReplicas` is used exclusively when `ReplicaStateMachine` is requested to start up.

`addLiveBrokersAndEpochs` Method

addLiveBrokersAndEpochs(
  brokerAndEpochs: Map[Broker, Long]): Unit

addLiveBrokersAndEpochs…FIXME

Note	`addLiveBrokersAndEpochs` is used when…FIXME

`removeLiveBrokers` Method

removeLiveBrokers(
  brokerIds: Set[Int]): Unit

removeLiveBrokers…FIXME

Note	`removeLiveBrokers` is used when…FIXME

`partitionReplicaAssignment` Method

partitionReplicaAssignment(
  topicPartition: TopicPartition): Seq[Int]

partitionReplicaAssignment finds the brokers with the replicas of the given partition (aka partition replica assignment).

Internally, partitionReplicaAssignment finds broker IDs of the replicas of the given partition (TopicPartition) in the partitionAssignments internal registry.

partitionReplicaAssignment returns an empty collection when no topic or partition are found.

Note	`partitionReplicaAssignment` is used when…FIXME

`putReplicaStateIfNotExists` Method

putReplicaStateIfNotExists(
  replica: PartitionAndReplica,
  state: ReplicaState): Unit

putReplicaStateIfNotExists simply adds the replica to the replicaStates internal registry unless available already.

Note	`putReplicaStateIfNotExists` is used exclusively when `ZkReplicaStateMachine` is requested to handle state changes of partition replicas.

`checkValidReplicaStateChange` Method

checkValidReplicaStateChange(
  replicas: Seq[PartitionAndReplica],
  targetState: ReplicaState
): (Seq[PartitionAndReplica], Seq[PartitionAndReplica])

For every replica (in the given replicas), checkValidReplicaStateChange isValidReplicaStateTransition with the target state (ReplicaState).

Note	`checkValidReplicaStateChange` is used exclusively when `ZkReplicaStateMachine` is requested to handle replica state changes.

`checkValidPartitionStateChange` Method

checkValidPartitionStateChange(
  partitions: Seq[TopicPartition],
  targetState: PartitionState
): (Seq[TopicPartition], Seq[TopicPartition])

For every replica (in the given replicas), checkValidPartitionStateChange isValidPartitionStateTransition with the target state (PartitionState).

Note	`checkValidPartitionStateChange` is used exclusively when `ZkReplicaStateMachine` is requested to handle partition state changes.

`putPartitionStateIfNotExists` Method

putPartitionStateIfNotExists(
  partition: TopicPartition,
  state: PartitionState): Unit

putPartitionStateIfNotExists…FIXME

Note	`putPartitionStateIfNotExists` is used when…FIXME

`isValidReplicaStateTransition` Internal Method

isValidReplicaStateTransition(
  replica: PartitionAndReplica,
  targetState: ReplicaState): Boolean

isValidReplicaStateTransition…FIXME

Note	`isValidReplicaStateTransition` is used when…FIXME

`isValidPartitionStateTransition` Internal Method

isValidPartitionStateTransition(
  partition: TopicPartition,
  targetState: PartitionState): Boolean

isValidPartitionStateTransition…FIXME

Note	`isValidPartitionStateTransition` is used when…FIXME

`putPartitionState` Method

putPartitionState(
  partition: TopicPartition,
  targetState: PartitionState): Unit

putPartitionState…FIXME

Note	`putPartitionState` is used when…FIXME

Finding Partitions by Given States — `partitionsInStates` Method

partitionsInStates(
  states: Set[PartitionState]): Set[TopicPartition]
partitionsInStates(
  topic: String, states: Set[PartitionState]): Set[TopicPartition]

partitionsInStates uses the partitionStates internal registry to find all of the TopicPartitions (of the topic if defined) in the given PartitionStates.

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

Internal Properties

Name Description

Name	Description
`allTopics`
`epoch`	Current controller epoch
`epochZkVersion`
`liveBrokerEpochs`	(Live and shutting down) Broker IDs and the current epoch `liveBrokerEpochs: Map[Int, Long]`
`liveBrokers`
`offlinePartitionCount`
`partitionAssignments`	Broker IDs of partition replicas (`Map[String, Map[Int, Seq[Int]]]`)
`partitionLeadershipInfo`	`partitionLeadershipInfo: mutable.Map[TopicPartition, LeaderIsrAndControllerEpoch]`
`partitionsBeingReassigned`
`partitionStates`
`replicasOnOfflineDirs`
`replicaStates`	`ReplicaStates` of `PartitionAndReplica` (`Map[PartitionAndReplica, ReplicaState]`)
`shuttingDownBrokerIds`
`stats`	`ControllerStats` with `UncleanLeaderElectionsPerSec` meter metric and `KafkaTimers` for every ControllerState (except Idle state) `stats` is used exclusively to create the ControllerEventManager (of KafkaController) that is then used to collect the times (metrics) of processing every ControllerEvent (except ShutdownEventThread) Every `ControllerState` has the RateAndTimeMs metric defined (except Idle state) The timer metric name pattern is kafka.controller:type=ControllerStats,name=.
`topicsIneligibleForDeletion`
`topicsToBeDeleted`
`topicsWithDeletionStarted`

allTopics

epoch

Current controller epoch

epochZkVersion

liveBrokerEpochs

(Live and shutting down) Broker IDs and the current epoch

liveBrokerEpochs: Map[Int, Long]

liveBrokers

offlinePartitionCount

partitionAssignments

Broker IDs of partition replicas (Map[String, Map[Int, Seq[Int]]])

partitionLeadershipInfo

partitionLeadershipInfo: mutable.Map[TopicPartition, LeaderIsrAndControllerEpoch]

partitionsBeingReassigned

partitionStates

replicasOnOfflineDirs

replicaStates

ReplicaStates of PartitionAndReplica (Map[PartitionAndReplica, ReplicaState])

shuttingDownBrokerIds

stats

ControllerStats with UncleanLeaderElectionsPerSec meter metric and KafkaTimers for every ControllerState (except Idle state)

stats is used exclusively to create the ControllerEventManager (of KafkaController) that is then used to collect the times (metrics) of processing every ControllerEvent (except ShutdownEventThread)

Every ControllerState has the RateAndTimeMs metric defined (except Idle state)

The timer metric name pattern is kafka.controller:type=ControllerStats,name=.

topicsIneligibleForDeletion

topicsToBeDeleted

topicsWithDeletionStarted

ControllerContext