AbstractControllerBrokerRequestBatch

AbstractControllerBrokerRequestBatch is the base of ControllerBrokerRequestBatches that sendRequestsToBrokers.

Note	ControllerBrokerRequestBatch is the default and only known implementation of the AbstractControllerBrokerRequestBatch Contract in Apache Kafka.

Table 1. AbstractControllerBrokerRequestBatch Contract (Abstract Methods Only)
Method	Description
`sendEvent`	`sendEvent( event: ControllerEvent): Unit` Sends a ControllerEvent Used when `AbstractControllerBrokerRequestBatch` is requested to sendLeaderAndIsrRequest and sendStopReplicaRequests
`sendRequest`	`sendRequest( brokerId: Int, request: AbstractControlRequest.Builder[_ <: AbstractControlRequest], callback: AbstractResponse => Unit = null): Unit` Sends an controller request out to the given broker with optional callback to handle a response Used when `AbstractControllerBrokerRequestBatch` is requested to sendLeaderAndIsrRequest, sendUpdateMetadataRequests, and sendStopReplicaRequests

Creating AbstractControllerBrokerRequestBatch Instance

AbstractControllerBrokerRequestBatch takes the following to be created:

AbstractControllerBrokerRequestBatch initializes the internal properties.

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

Sending Controller Requests To Brokers — `sendRequestsToBrokers` Method

sendRequestsToBrokers(
  controllerEpoch: Int): Unit

sendRequestsToBrokers sends LeaderAndIsr requests out to brokers.

sendRequestsToBrokers sends UpdateMetadata requests out to brokers.

sendRequestsToBrokers sends StopReplica requests out to brokers.

In case of any error (Throwable), sendRequestsToBrokers…FIXME

Note	`sendRequestsToBrokers` is used when: `KafkaController` is requested to updateLeaderEpochAndSendRequest, sendUpdateMetadataRequest and at ControlledShutdown controller event `ZkPartitionStateMachine` is requested to handleStateChanges `ReplicaStateMachine` is requested to handleStateChanges

Sending LeaderAndIsr Requests Out to All Brokers — `sendLeaderAndIsrRequest` Internal Method

sendLeaderAndIsrRequest(
  controllerEpoch: Int,
  stateChangeLog: StateChangeLogger): Unit

sendLeaderAndIsrRequest…FIXME

Note	`sendLeaderAndIsrRequest` is used exclusively when `AbstractControllerBrokerRequestBatch` is requested to sendRequestsToBrokers.

Sending UpdateMetadata Requests Out to All Brokers — `sendUpdateMetadataRequests` Internal Method

sendUpdateMetadataRequests(
  controllerEpoch: Int,
  stateChangeLog: StateChangeLogger): Unit

sendUpdateMetadataRequests prints out the following TRACE message to the logs for every pair in the updateMetadataRequestPartitionInfoMap internal registry:

Sending UpdateMetadata request [partitionState] to brokers [updateMetadataRequestBrokerSet] for partition [tp]

sendUpdateMetadataRequests prepares metadata for UpdateMetadataRequests.

For every broker (in the updateMetadataRequestBrokerSet internal registry) that is live or shutting down sendUpdateMetadataRequests requests the ControllerContext for the broker epoch, creates a new UpdateMetadataRequest (with the updateMetadataRequestPartitionInfoMap among others) and then sends the request out to the broker.

In the end, sendUpdateMetadataRequests removes all elements from (clears) the updateMetadataRequestBrokerSet and updateMetadataRequestPartitionInfoMap internal registries.

Note	`sendUpdateMetadataRequests` is used exclusively when `AbstractControllerBrokerRequestBatch` is requested to send controller requests to all brokers in a cluster.

Sending StopReplica Requests Out to All Brokers — `sendStopReplicaRequests` Internal Method

sendStopReplicaRequests(
  controllerEpoch: Int): Unit

sendStopReplicaRequests…FIXME

Note	`sendStopReplicaRequests` is used exclusively when `AbstractControllerBrokerRequestBatch` is requested to sendRequestsToBrokers.

Preparing New Batch — `newBatch` Method

newBatch(): Unit

newBatch simply throws an IllegalStateException when any of the following internal registries are not empty: leaderAndIsrRequestMap, stopReplicaRequestMap and updateMetadataRequestBrokerSet.

Controller to broker state change requests batch is not empty while creating a new one. Some LeaderAndIsr state changes [leaderAndIsrRequestMap] might be lost

Controller to broker state change requests batch is not empty while creating a new one. Some StopReplica state changes [stopReplicaRequestMap] might be lost

Controller to broker state change requests batch is not empty while creating a new one. Some UpdateMetadata state changes to brokers [updateMetadataRequestBrokerSet] with partition info [updateMetadataRequestPartitionInfoMap] might be lost

Note	`newBatch` is used when: `KafkaController` is requested to updateLeaderEpochAndSendRequest, sendUpdateMetadataRequest, and process a ControlledShutdown controller event `PartitionStateMachine` is requested to handleStateChanges `PartitionStateMachine` is requested to handleStateChanges

`addLeaderAndIsrRequestForBrokers` Method

addLeaderAndIsrRequestForBrokers(
  brokerIds: Seq[Int],
  topicPartition: TopicPartition,
  leaderIsrAndControllerEpoch: LeaderIsrAndControllerEpoch,
  replicas: Seq[Int],
  isNew: Boolean): Unit

addLeaderAndIsrRequestForBrokers looks up the LeaderAndIsrPartitionStates of TopicPartitions on a broker in the leaderAndIsrRequestMap internal registry. addLeaderAndIsrRequestForBrokers uses IDs from the given brokerIds that are 0 or higher.

addLeaderAndIsrRequestForBrokers adds (or replaces) the given topicPartition in the result with a new LeaderAndIsrPartitionState based on the input arguments.

In the end, addLeaderAndIsrRequestForBrokers addUpdateMetadataRequestForBrokers to the live or shutting down brokers for the requested partition (topicPartition).

Note	`addLeaderAndIsrRequestForBrokers` is used when: `KafkaController` is requested to updateLeaderEpochAndSendRequest `ZkPartitionStateMachine` is requested to initializeLeaderAndIsrForPartitions and doElectLeaderForPartitions `ZkReplicaStateMachine` is requested to \

`addUpdateMetadataRequestForBrokers` Method

addUpdateMetadataRequestForBrokers(
  brokerIds: Seq[Int],
  partitions: collection.Set[TopicPartition]): Unit

addUpdateMetadataRequestForBrokers…FIXME

Note	`addUpdateMetadataRequestForBrokers` is used when: `AbstractControllerBrokerRequestBatch` is requested to addLeaderAndIsrRequestForBrokers `KafkaController` is requested to sendUpdateMetadataRequest `ZkReplicaStateMachine` is requested to doHandleStateChanges

Internal Properties

Name Description

Name	Description
`leaderAndIsrRequestMap`	`LeaderAndIsrPartitionState` of `TopicPartitions` by broker ID `leaderAndIsrRequestMap: Map[ Int, Map[TopicPartition, LeaderAndIsrRequestData.LeaderAndIsrPartitionState]]` Used when…FIXME
`stopReplicaRequestMap`	(`Map[Int, ListBuffer[StopReplicaRequestInfo]]`) Used when…FIXME
`updateMetadataRequestBrokerSet`	Broker IDs to send UpdateMetadata requests out to Broker IDs added in addUpdateMetadataRequestForBrokers Cleared (emptied) in clear and after successful sendUpdateMetadataRequests

leaderAndIsrRequestMap

LeaderAndIsrPartitionState of TopicPartitions by broker ID

leaderAndIsrRequestMap: Map[
  Int,
  Map[TopicPartition, LeaderAndIsrRequestData.LeaderAndIsrPartitionState]]

Used when…FIXME

stopReplicaRequestMap

(Map[Int, ListBuffer[StopReplicaRequestInfo]])

Used when…FIXME

updateMetadataRequestBrokerSet

Broker IDs to send UpdateMetadata requests out to

Broker IDs added in addUpdateMetadataRequestForBrokers
Cleared (emptied) in clear and after successful sendUpdateMetadataRequests

AbstractControllerBrokerRequestBatch