AbstractFetcherManager

AbstractFetcherManager is the abstraction of fetcher thread managers that can create fetcher threads (for ReplicaManager).

Note	`AbstractFetcherManager` is an abstract class and cannot be created directly. It is created indirectly for the concrete AbstractFetcherManagers.

Table 1. AbstractFetcherManager Contract (Abstract Methods)
Method	Description
`createFetcherThread`	`createFetcherThread( fetcherId: Int, sourceBroker: BrokerEndPoint): T` Creates a fetcher thread Used when `AbstractFetcherManager` is requested to addFetcherForPartitions

Table 2. AbstractFetcherManagers
AbstractFetcherManager	Description
ReplicaAlterLogDirsManager
ReplicaFetcherManager

Performance Metrics

AbstractFetcherManagers are KafkaMetricsGroups with the following performance metrics.

Table 3. AbstractFetcherManager’s Performance Metrics
Metric Name	Description
`DeadThreadCount`
`FailedPartitionsCount`
`MaxLag`
`MinFetchRate`

The performance metrics are registered in kafka.server:type= group with the type being the name of a AbstractFetcherManager.

Creating AbstractFetcherManager Instance

AbstractFetcherManager takes the following to be created:

Log Prefix
Client ID (for a tag in performance metrics)
num.replica.fetchers

AbstractFetcherManager initializes the internal properties.

`addFetcherForPartitions` Method

addFetcherForPartitions(
  partitionAndOffsets: Map[TopicPartition, InitialFetchState]): Unit

addFetcherForPartitions…FIXME

Note	`addFetcherForPartitions` is used when: `AbstractFetcherManager` is requested to resizeThreadPool (when `DynamicThreadPool` is requested to reconfigure) `ReplicaManager` is requested to alterReplicaLogDirs, becomeLeaderOrFollower and makeFollowers

`resizeThreadPool` Method

resizeThreadPool(newSize: Int): Unit

resizeThreadPool prints out the following INFO message to the logs:

Resizing fetcher thread pool size from [currentSize] to [newSize]

resizeThreadPool…FIXME

Note	`resizeThreadPool` is used exclusively when `DynamicThreadPool` is requested to reconfigure.

Finding Fetcher Thread ID — `getFetcherId` Method

getFetcherId(
  topicPartition: TopicPartition): Int

getFetcherId calculates a hash of the topic and partition (from the given TopicPartition) modulo (%) the numFetchersPerBroker.

Note	`getFetcherId` is used when `AbstractFetcherManager` is requested to markPartitionsForTruncation and addFetcherForPartitions.

`getFetcher` Method

getFetcher(topicPartition: TopicPartition): Option[T]

getFetcher…FIXME

Note	`getFetcher` is used when…FIXME

`markPartitionsForTruncation` Method

markPartitionsForTruncation(
  brokerId: Int,
  topicPartition: TopicPartition,
  truncationOffset: Long): Unit

markPartitionsForTruncation…FIXME

Note	`markPartitionsForTruncation` is used when…FIXME

`removeFetcherForPartitions` Method

removeFetcherForPartitions(
  partitions: Set[TopicPartition]): Unit

removeFetcherForPartitions…FIXME

Note	`removeFetcherForPartitions` is used when: `AbstractFetcherManager` is requested to resizeThreadPool `ReplicaManager` is requested to stopReplicas, alterReplicaLogDirs, makeLeaders, makeFollowers, and handleLogDirFailure

`closeAllFetchers` Method

closeAllFetchers(): Unit

closeAllFetchers requests the fetcher threads to initiateShutdown.

closeAllFetchers requests the fetcher threads to shutdown.

In the end, closeAllFetchers removes (clears) all entries from the fetcherThreadMap internal registry.

Note	`closeAllFetchers` is used when ReplicaAlterLogDirsManager and ReplicaFetcherManager are requested to shut down.

Internal Properties

Name Description

Name	Description
`fetcherThreadMap`	AbstractFetcherThreads by (a pair of) broker ID and fetcher ID
`numFetchersPerBroker`	Number of fetcher threads per broker Initialized with the given numFetchers when `AbstractFetcherManager` is created Resized in resizeThreadPool Used in getFetcherId

fetcherThreadMap

AbstractFetcherThreads by (a pair of) broker ID and fetcher ID

numFetchersPerBroker

Number of fetcher threads per broker

Initialized with the given numFetchers when AbstractFetcherManager is created

Resized in resizeThreadPool

Used in getFetcherId

AbstractFetcherManager