kafka-server-ReplicaManager.adoc

ReplicaManager

ReplicaManager manages log replicas using the LogManager.

ReplicaManager is created and immediately started when KafkaServer is requested to start up.

Figure 1. ReplicaManager and KafkaServer

ReplicaManager is given all the required services (e.g. Metrics, KafkaZkClient, Scheduler, LogManager, QuotaManagers, MetadataCache) from the KafkaServer.

Note	KafkaServer creates a ReplicaManager for the only purpose of creating KafkaApis, GroupCoordinator, and TransactionCoordinator.

When started, ReplicaManager schedules isr-expiration and isr-change-propagation recurring tasks (every half of replica.lag.time.max.ms property and 2500 ms, respectively).

ReplicaManager manages a registry of Partitions by TopicPartition.

ReplicaManager can be the leader or a follower of partitions which happens when KafkaApis is requested to handle a LeaderAndIsr request (mostly, if not always, from the KafkaController).

ReplicaManager uses the MetadataCache for the following:

• …​FIXME

Tip

Enable ALL logging level for kafka.server.ReplicaManager logger to see what happens inside. Add the following line to log4j.properties: log4j.logger.kafka.server.ReplicaManager=ALL Refer to Logging. Please note that ReplicaManager also uses a preconfigured state.change.logger logger in config/log4j.properties: log4j.appender.stateChangeAppender=org.apache.log4j.DailyRollingFileAppender log4j.appender.stateChangeAppender.DatePattern='.'yyyy-MM-dd-HH log4j.appender.stateChangeAppender.File=${kafka.logs.dir}/state-change.log log4j.appender.stateChangeAppender.layout=org.apache.log4j.PatternLayout log4j.appender.stateChangeAppender.layout.ConversionPattern=[%d] %p %m (%c)%n log4j.logger.state.change.logger=TRACE, stateChangeAppender log4j.additivity.state.change.logger=false That means that the logs of ReplicaManager go to logs/state-change.log file at TRACE logging level and are not added to the main logs (per log4j.additivity being off).

ReplicaManager and ReplicaFetcherManager

ReplicaManager creates a ReplicaFetcherManager while being created (via createReplicaFetcherManager).

Figure 2. ReplicaManager and ReplicaFetcherManager

ReplicaManager uses the ReplicaFetcherManager for the following:

• addFetcherForPartitions when makeFollowers

• removeFetcherForPartitions when stopReplicas, makeLeaders, makeFollowers and handleLogDirFailure

• shutdownIdleFetcherThreads when becomeLeaderOrFollower

• shutdown when shutdown

ReplicaFetcherManager is also used when:

• DynamicThreadPool is requested to reconfigure (resize) the thread pool

• KafkaApis is requested to handle a StopReplica request

Performance Metrics

ReplicaManager is a KafkaMetricsGroup with the following performance metrics.

Table 1. ReplicaManager’s Performance Metrics

Metric Name	Description
FailedIsrUpdatesPerSec
IsrExpandsPerSec
IsrShrinksPerSec
LeaderCount
OfflineReplicaCount
PartitionCount	Number of allPartitions
UnderMinIsrPartitionCount
UnderReplicatedPartitions	underReplicatedPartitionCount

The performance metrics are registered in kafka.server:type=ReplicaManager group.

Figure 3. ReplicaManager in jconsole

Registry of Partitions by TopicPartition — allPartitions Internal Property

allPartitions: Pool[TopicPartition, Partition]

allPartitions is a registry of Partitions by TopicPartition

The number of partitions in allPartitions is available as the PartitionCount performance metric.

A new TopicPartition is added exclusively when looking up a Partition by a TopicPartition that is not registered yet.

A TopicPartition is removed when stopping a replica (and marked as the OfflinePartition).

Note	allPartitions is used when getPartition, nonOfflinePartitionsIterator, offlinePartitionsIterator, becomeLeaderOrFollower, maybeShrinkIsr, handleLogDirFailure

Creating ReplicaFetcherManager — createReplicaFetcherManager Internal Method

createReplicaFetcherManager(
  metrics: Metrics
  time: Time
  threadNamePrefix: Option[String]
  quotaManager: ReplicationQuotaManager): ReplicaFetcherManager

createReplicaFetcherManager simply creates a ReplicaFetcherManager.

Note	createReplicaFetcherManager is used when ReplicaManager is created.

Creating ReplicaAlterLogDirsManager — createReplicaAlterLogDirsManager Internal Method

createReplicaAlterLogDirsManager(
  quotaManager: ReplicationQuotaManager,
  brokerTopicStats: BrokerTopicStats): ReplicaAlterLogDirsManager

createReplicaAlterLogDirsManager simply creates a ReplicaAlterLogDirsManager.

Note	createReplicaAlterLogDirsManager is used when ReplicaManager is created.

shutdown Method

shutdown(
  checkpointHW: Boolean = true): Unit

shutdown…​FIXME

Note	shutdown is used when KafkaServer is requested to shut down.

alterReplicaLogDirs Method

alterReplicaLogDirs(partitionDirs: Map[TopicPartition, String]): Map[TopicPartition, Errors]

alterReplicaLogDirs…​FIXME

Note	alterReplicaLogDirs is used exclusively when KafkaApis is requested to handle an AlterReplicaLogDirs request.

Becoming Leader or Follower of Partitions — becomeLeaderOrFollower Method

becomeLeaderOrFollower(
  correlationId: Int,
  leaderAndIsrRequest: LeaderAndIsrRequest,
  onLeadershipChange: (Iterable[Partition], Iterable[Partition]) => Unit): LeaderAndIsrResponse

becomeLeaderOrFollower prints out the following TRACE message to the logs:

Received LeaderAndIsr request [stateInfo] correlation id [correlationId] from controller [controllerId] epoch [controllerEpoch] for partition [topicPartition]

becomeLeaderOrFollower records the current controller epoch (of the LeaderAndIsrRequest) in the controllerEpoch internal registry.

For all valid partition states, becomeLeaderOrFollower finds the partition states with the leader being the (local) broker and that are the partitions for which the broker becomes the leader. All other partition states are for partitions for which the broker becomes a follower.

For all partitions for which the broker becomes the leader, becomeLeaderOrFollower makeLeaders.

For all partitions for which the broker becomes a follower, becomeLeaderOrFollower makeFollowers.

With the hwThreadInitialized internal flag disabled (false), becomeLeaderOrFollower startHighWaterMarksCheckPointThread and turns the flag on (true).

For every new partitions, becomeLeaderOrFollower…​FIXME

becomeLeaderOrFollower…​FIXME

becomeLeaderOrFollower calls the given onLeadershipChange callback with the partitions for the broker to be the leader and a follower.

In the end, becomeLeaderOrFollower creates a new LeaderAndIsrResponse to "announce" a successful request processing.

Note	becomeLeaderOrFollower is used exclusively when KafkaApis is requested to handle a LeaderAndIsr request.

makeFollowers Internal Method

makeFollowers(
  controllerId: Int,
  epoch: Int,
  partitionState: Map[Partition, LeaderAndIsrRequest.PartitionState],
  correlationId: Int,
  responseMap: mutable.Map[TopicPartition, Errors]) : Set[Partition]

makeFollowers…​FIXME

Note	makeFollowers is used when ReplicaManager is requested to becomeLeaderOrFollower.

recordIsrChange Method

recordIsrChange(topicPartition: TopicPartition): Unit

recordIsrChange adds the input topicPartition to isrChangeSet internal registry and sets lastIsrChangeMs to the current time.

Note	recordIsrChange is used exclusively when Partition does updateIsr

updateFollowerLogReadResults Internal Method

updateFollowerLogReadResults(
  replicaId: Int,
  readResults: Seq[(TopicPartition, LogReadResult)]): Seq[(TopicPartition, LogReadResult)]

updateFollowerLogReadResults…​FIXME

Note	updateFollowerLogReadResults is used exclusively when ReplicaManager is requested to fetch messages from the leader replica.

fetchMessages Method

fetchMessages(
  timeout: Long,
  replicaId: Int,
  fetchMinBytes: Int,
  fetchMaxBytes: Int,
  hardMaxBytesLimit: Boolean,
  fetchInfos: Seq[(TopicPartition, PartitionData)],
  quota: ReplicaQuota,
  responseCallback: Seq[(TopicPartition, FetchPartitionData)] => Unit,
  isolationLevel: IsolationLevel,
  clientMetadata: Option[ClientMetadata]): Unit

fetchMessages determines fetchIsolation:

• FetchLogEnd for a request from a follower or when the replicaId is Request.FutureLocalReplicaId

• FetchTxnCommitted for isolationLevel being READ_COMMITTED

• FetchHighWatermark otherwise

fetchMessages readFromLocalLog (with the fetchIsolation).

fetchMessages…​FIXME

Note	fetchMessages is used when: KafkaApis is requested to handle a Fetch request ReplicaAlterLogDirsThread is requested to fetchFromLeader

maybePropagateIsrChanges Method

maybePropagateIsrChanges(): Unit

maybePropagateIsrChanges…​FIXME

Note	maybePropagateIsrChanges is used exclusively when isr-change-propagation task is executed (every 2500 milliseconds).

Creating ReplicaManager Instance

ReplicaManager takes the following when created:

• KafkaConfig

• Metrics

• Time

• KafkaZkClient

• Scheduler

• LogManager

• isShuttingDown flag

• QuotaManagers

• BrokerTopicStats

• MetadataCache

• LogDirFailureChannel

• DelayedOperationPurgatory[DelayedProduce]

• DelayedOperationPurgatory[DelayedFetch]

• DelayedOperationPurgatory[DelayedDeleteRecords]

• Optional thread name prefix

ReplicaManager initializes the internal registries and counters.

Starting ReplicaManager (and Scheduling ISR-Related Tasks) — startup Method

startup(): Unit

startup requests Scheduler to schedule the ISR-related tasks:

1. isr-expiration

2. isr-change-propagation

startup then creates a LogDirFailureHandler and requests it to start.

Note	startup uses Scheduler that was specified when ReplicaManager was created.

Note	startup is used exclusively when KafkaServer starts up.

maybeShrinkIsr Internal Method

maybeShrinkIsr(): Unit

maybeShrinkIsr prints out the following TRACE message to the logs:

Evaluating ISR list of partitions to see which replicas can be removed from the ISR

maybeShrinkIsr requests the partitions (from allPartitions pool that are not offline partitions) to maybeShrinkIsr (with replica.lag.time.max.ms property).

Note	maybeShrinkIsr is used exclusively to schedule isr-expiration recurring task when ReplicaManager starts up.

makeLeaders Internal Method

makeLeaders(
  controllerId: Int,
  epoch: Int,
  partitionState: Map[Partition, LeaderAndIsrRequest.PartitionState],
  correlationId: Int,
  responseMap: mutable.Map[TopicPartition, Errors]): Set[Partition]

makeLeaders…​FIXME

Note	makeLeaders is used exclusively when ReplicaManager is requested to becomeLeaderOrFollower.

describeLogDirs Method

describeLogDirs(partitions: Set[TopicPartition]): Map[String, LogDirInfo]

describeLogDirs…​FIXME

Note	describeLogDirs is used exclusively when KafkaApis is requested to handle a DescribeLogDirs request.

Finding Log For TopicPartition — getLog Method

getLog(topicPartition: TopicPartition): Option[Log]

getLog…​FIXME

Note	getLog is used when: GroupMetadataManager is requested to doLoadGroupsAndOffsets TransactionStateManager is requested to loadTransactionMetadata

startHighWaterMarksCheckPointThread Method

startHighWaterMarksCheckPointThread(): Unit

startHighWaterMarksCheckPointThread…​FIXME

Note	startHighWaterMarksCheckPointThread is used when…​FIXME

checkpointHighWatermarks Method

checkpointHighWatermarks(): Unit

checkpointHighWatermarks…​FIXME

Note	checkpointHighWatermarks is used when…​FIXME

shutdownIdleReplicaAlterLogDirsThread Method

shutdownIdleReplicaAlterLogDirsThread(): Unit

shutdownIdleReplicaAlterLogDirsThread…​FIXME

Note	shutdownIdleReplicaAlterLogDirsThread is used when…​FIXME

handleLogDirFailure Method

handleLogDirFailure(
  dir: String,
  sendZkNotification: Boolean = true): Unit

handleLogDirFailure…​FIXME

Note	handleLogDirFailure is used when LogDirFailureHandler is requested to do the work.

maybeUpdateMetadataCache Method

maybeUpdateMetadataCache(
  correlationId: Int,
  updateMetadataRequest: UpdateMetadataRequest) : Seq[TopicPartition]

maybeUpdateMetadataCache…​FIXME

Note	maybeUpdateMetadataCache is used exclusively when KafkaApis is requested to handle an UpdateMetadata request.

Appending Records — appendRecords Method

appendRecords(
  timeout: Long,
  requiredAcks: Short,
  internalTopicsAllowed: Boolean,
  isFromClient: Boolean,
  entriesPerPartition: Map[TopicPartition, MemoryRecords],
  responseCallback: Map[TopicPartition, PartitionResponse] => Unit,
  delayedProduceLock: Option[Lock] = None,
  recordConversionStatsCallback: Map[TopicPartition, RecordConversionStats] => Unit = _ => ()): Unit

appendRecords…​FIXME

Note	appendRecords is used when: GroupMetadataManager is requested to request the ReplicaManager to append records TransactionStateManager is requested to enableTransactionalIdExpiration and appendTransactionToLog KafkaApis is requested to handle Produce and WriteTxnMarkers requests

Validating requiredAcks — isValidRequiredAcks Internal Method

isValidRequiredAcks(requiredAcks: Short): Boolean

isValidRequiredAcks is positive (true) when the given requiredAcks is one of the following:

• -1

• 1

• 0

Otherwise, isValidRequiredAcks is negative (false).

Note	isValidRequiredAcks is used exclusively when ReplicaManager is requested to appendRecords.

appendToLocalLog Internal Method

appendToLocalLog(
  internalTopicsAllowed: Boolean,
  isFromClient: Boolean,
  entriesPerPartition: Map[TopicPartition, MemoryRecords],
  requiredAcks: Short): Map[TopicPartition, LogAppendResult]

appendToLocalLog processes (maps over) the given Map[TopicPartition, MemoryRecords] (entriesPerPartition), so that the leader partition (of every TopicPartition) is requested to appendRecordsToLeader.

Internally, appendToLocalLog prints out the following TRACE message to the logs:

Append [[entriesPerPartition]] to local log

For every tuple in the given entriesPerPartition (Map[TopicPartition, MemoryRecords]), appendToLocalLog does the following steps:

1. Requests the BrokerTopicStats to mark the occurrence of an event for the totalProduceRequestRate for the topic (of the TopicPartition) in the topicStats and for all topics

2. Gets the partition (or throws an exception) (with expectLeader flag enabled)

3. Requests the Partition to appendRecordsToLeader (with the MemoryRecords, the isFromClient flag, and the requiredAcks bit map)

4. Requests the BrokerTopicStats to mark the sizeInBytes of the MemoryRecords for the bytesInRate for the topic (of the TopicPartition) in the bytesInRate and for all topics

5. Requests the BrokerTopicStats to mark the number of messages appended for the messagesInRate for the topic (of the TopicPartition) in the bytesInRate and for all topics

6. Prints out the following TRACE message to the logs:

   [sizeInBytes] written to log [topicPartition] beginning at offset [firstOffset] and ending at offset [lastOffset]

In case Topic.isInternal(topicPartition.topic) && !internalTopicsAllowed, appendToLocalLog…​FIXME

In case of exceptions, appendToLocalLog…​FIXME

Note	appendToLocalLog is used exclusively when ReplicaManager is requested to append records.

Getting Partition Or Throwing Exception — getPartitionOrException Method

getPartitionOrException(
  topicPartition: TopicPartition,
  expectLeader: Boolean): Partition

getPartitionOrException gets the partition if available or throws one of the following exceptions:

• KafkaStorageException when the partition is offline

  Partition [topicPartition] is in an offline log directory

• NotLeaderForPartitionException

  Broker [localBrokerId] is not a replica of [topicPartition]

• ReplicaNotAvailableException

  Partition [topicPartition] is not available

• UnknownTopicOrPartitionException

  Partition [topicPartition] doesn't exist

Note	getPartitionOrException is used when…​FIXME

Getting Partition by TopicPartition (If Available) — getPartition Method

getPartition(topicPartition: TopicPartition): Option[Partition]

getPartition gets the partition for the given TopicPartition.

Note

getPartition is used when: DelayedDeleteRecords is requested to tryComplete DelayedProduce is requested to tryComplete ReplicaAlterLogDirsThread is requested to processPartitionData ReplicaFetcherThread is requested to processPartitionData, truncate, and truncateFullyAndStartAt ReplicaManager is requested to nonOfflinePartition, getPartitionOrException, alterReplicaLogDirs, appendToLocalLog, becomeLeaderOrFollower, and lastOffsetForLeaderEpoch

Stopping Partition Replica — stopReplica Method

stopReplica(
  topicPartition: TopicPartition,
  deletePartition: Boolean): Unit

stopReplica…​FIXME

Note	stopReplica is used exclusively when ReplicaManager is requested to stopReplicas.

underReplicatedPartitionCount Method

underReplicatedPartitionCount: Int

underReplicatedPartitionCount…​FIXME

Note	underReplicatedPartitionCount is used exclusively for the UnderReplicatedPartitions performance metric.

leaderPartitionsIterator Internal Method

leaderPartitionsIterator: Iterator[Partition]

leaderPartitionsIterator…​FIXME

Note	leaderPartitionsIterator is used exclusively for the performance metrics: LeaderCount, UnderMinIsrPartitionCount, and UnderReplicatedPartitions (indirectly using underReplicatedPartitionCount).

nonOfflinePartitionsIterator Internal Method

nonOfflinePartitionsIterator: Iterator[Partition]

nonOfflinePartitionsIterator…​FIXME

Note	nonOfflinePartitionsIterator is used when ReplicaManager is requested to leaderPartitionsIterator, checkpointHighWatermarks, and handleLogDirFailure.

Looking Up Partition or Creating New One (by TopicPartition) — getOrCreatePartition Method

getOrCreatePartition(topicPartition: TopicPartition): Partition

getOrCreatePartition simply looks up a Partition by the TopicPartition (in the allPartitions internal registry). If not found, getOrCreatePartition adds a new Partition.

Note	getOrCreatePartition is used exclusively when ReplicaManager is requested to becomeLeaderOrFollower.

offlinePartitionsIterator Internal Method

offlinePartitionsIterator: Iterator[Partition]

offlinePartitionsIterator…​FIXME

Note	offlinePartitionsIterator is used when…​FIXME

markPartitionOffline Method

markPartitionOffline(tp: TopicPartition): Unit

markPartitionOffline…​FIXME

Note	markPartitionOffline is used when…​FIXME

lastOffsetForLeaderEpoch Method

lastOffsetForLeaderEpoch(
  requestedEpochInfo: Map[TopicPartition, OffsetsForLeaderEpochRequest.PartitionData]
): Map[TopicPartition, EpochEndOffset]

lastOffsetForLeaderEpoch…​FIXME

Note	lastOffsetForLeaderEpoch is used when…​FIXME

nonOfflinePartition Method

nonOfflinePartition(topicPartition: TopicPartition): Option[Partition]

nonOfflinePartition…​FIXME

Note	nonOfflinePartition is used when…​FIXME

deleteRecords Method

deleteRecords(
  timeout: Long,
  offsetPerPartition: Map[TopicPartition, Long],
  responseCallback: Map[TopicPartition, DeleteRecordsResponse.PartitionResponse] => Unit): Unit

deleteRecords…​FIXME

Note	deleteRecords is used when…​FIXME

fetchOffsetForTimestamp Method

fetchOffsetForTimestamp(
  topicPartition: TopicPartition,
  timestamp: Long,
  isolationLevel: Option[IsolationLevel],
  currentLeaderEpoch: Optional[Integer],
  fetchOnlyFromLeader: Boolean): TimestampOffset

fetchOffsetForTimestamp…​FIXME

Note	fetchOffsetForTimestamp is used when…​FIXME

stopReplicas Method

stopReplicas(
  stopReplicaRequest: StopReplicaRequest): (mutable.Map[TopicPartition, Errors], Errors)

stopReplicas…​FIXME

Note	stopReplicas is used exclusively when KafkaApis is requested to handle a StopReplica request.

getMagic Method

getMagic(topicPartition: TopicPartition): Option[Byte]

getMagic…​FIXME

Note	getMagic is used when…​FIXME

localReplicaOrException Method

localReplicaOrException(topicPartition: TopicPartition): Replica

localReplicaOrException finds the partition (or throws an exception) for the given TopicPartition (and expectLeader flag off) and requests the Partition to get the local partition replica (or throw an exception).

Note	A partition replica is local when the replica ID is exactly the local broker ID.

Note	localReplicaOrException is used when…​FIXME

shouldLeaderThrottle Method

shouldLeaderThrottle(
  quota: ReplicaQuota,
  topicPartition: TopicPartition,
  replicaId: Int): Boolean

shouldLeaderThrottle…​FIXME

Note	shouldLeaderThrottle is used when…​FIXME

readFromLocalLog Method

readFromLocalLog(
  replicaId: Int,
  fetchOnlyFromLeader: Boolean,
  fetchIsolation: FetchIsolation,
  fetchMaxBytes: Int,
  hardMaxBytesLimit: Boolean,
  readPartitionInfo: Seq[(TopicPartition, PartitionData)],
  quota: ReplicaQuota,
  clientMetadata: Option[ClientMetadata]): Seq[(TopicPartition, LogReadResult)]

readFromLocalLog returns records (as FetchDataInfo) for every TopicPartition in the given readPartitionInfo.

Internally, readFromLocalLog reads the requested PartitionData (up to fetchMaxBytes that is decremented sequentially for every partition) for every TopicPartition (in the given readPartitionInfo collection).

Note	readFromLocalLog is used when: DelayedFetch is requested to onComplete ReplicaManager is requested to fetchMessages (when KafkaApis is requested to handle a Fetch request)

read Internal Method

read(
  tp: TopicPartition,
  fetchInfo: PartitionData,
  limitBytes: Int,
  minOneMessage: Boolean): LogReadResult

read increments the totalFetchRequestRate metric of the topic (of the TopicPartition) and the allTopicsStats in the BrokerTopicStats.

read prints out the following TRACE message to the logs:

Fetching log segment for partition [tp], offset [offset], partition fetch size [partitionFetchSize], remaining response limit [limitBytes]

read finds the Partition for the TopicPartition.

read…​FIXME

Triggerring Preferred Replica Leader Election — electPreferredLeaders Method

electPreferredLeaders(
  controller: KafkaController,
  partitions: Set[TopicPartition],
  responseCallback: Map[TopicPartition, ApiError] => Unit,
  requestTimeout: Long): Unit

electPreferredLeaders simply requests the KafkaController to electPreferredLeaders for the partitions.

Note	electPreferredLeaders is used exclusively when KafkaApis is requested to handle an ElectPreferredLeaders request.

tryCompleteElection Method

tryCompleteElection(key: DelayedOperationKey): Unit

tryCompleteElection…​FIXME

Note	tryCompleteElection is used exclusively when KafkaApis is requested to handle an UpdateMetadata request.

localLog Method

localLog(
  topicPartition: TopicPartition): Option[Log]

localLog…​FIXME

Note	localLog is used when…​FIXME

findPreferredReadReplica Method

findPreferredReadReplica(
  tp: TopicPartition,
  clientMetadata: ClientMetadata,
  replicaId: Int,
  fetchOffset: Long,
  currentTimeMs: Long): Option[Int]

findPreferredReadReplica finds the Partition for the TopicPartition (with expectLeader flag off).

findPreferredReadReplica…​FIXME

Note	findPreferredReadReplica is used when ReplicaManager is requested to readFromLocalLog.

localLogOrException Method

localLogOrException(
  topicPartition: TopicPartition): Log

localLogOrException finds the Partition for the given TopicPartition (and expectLeader flag off) and then requests it for the local Log.

localLogOrException throws an exception if neither the Partition nor the Log of the given TopicPartition is available locally.

Note	localLogOrException is used when ReplicaFetcherThread is requested for the latest epoch, logEndOffset, endOffsetForEpoch, and buildFetch.

updateFollowerFetchState Internal Method

updateFollowerFetchState(
  followerId: Int,
  readResults: Seq[(TopicPartition, LogReadResult)]): Seq[(TopicPartition, LogReadResult)]

updateFollowerFetchState…​FIXME

Note	updateFollowerFetchState is used when ReplicaManager is requested to fetchMessages.

Internal Properties

Name	Description
controllerEpoch
replicaAlterLogDirsManager	ReplicaAlterLogDirsManager Created immediately with ReplicaManager Used when: ReplicaFetcherThread is requested to truncate ReplicaManager is requested to shutdownIdleReplicaAlterLogDirsThread, stopReplicas, alterReplicaLogDirs, becomeLeaderOrFollower, handleLogDirFailure, shutdown
highWatermarkCheckpoints	OffsetCheckpointFiles per live log data directory
stateChangeLogger	StateChangeLogger with the broker ID and inControllerContext flag off (that sets Broker log prefix)
hwThreadInitialized
isrChangeSet	Collection of TopicPartition that…​FIXME
lastIsrChangeMs	Time when isrChangeSet has a new TopicPartition added.
logDirFailureHandler	LogDirFailureHandler
OfflinePartition