KafkaRequestHandlerPool — Pool of KafkaRequestHandler Daemon Threads

KafkaRequestHandlerPool is a pool of KafkaRequestHandler daemon threads.

KafkaRequestHandlerPool is created exclusively when KafkaServer is requested to start up (and creates the data-plane and control-plane request handler pools).

Figure 1. KafkaRequestHandlerPool and KafkaRequestHandler Threads

When created, KafkaRequestHandlerPool requests all KafkaRequestHandler daemon threads to start.

Note	The number of kafka-request-handler threads is controlled by num.network.threads configuration property (default: `3`).

KafkaRequestHandlerPool uses [Kafka Request Handler on Broker [brokerId]] as the logging prefix (aka logIdent).

Tip	Enable `ALL` logging level for `kafka.server.KafkaRequestHandlerPool` logger to see what happens inside. Add the following line to `config/log4j.properties`: `log4j.logger.kafka.server.KafkaRequestHandlerPool=ALL` Refer to Logging.

Creating KafkaRequestHandlerPool Instance

KafkaRequestHandlerPool takes the following to be created:

Broker ID
RequestChannel
KafkaApis
Time
Number of threads (i.e. instances of KafkaRequestHandlers as defined by numNetworkThreads property)
requestHandlerAvgIdleMetricName
logAndThreadNamePrefix

KafkaRequestHandlerPool initializes the internal properties.

KafkaRequestHandlerPool starts numThreads daemon kafka-request-handler threads (as registered in the runnables internal registry).

Performance Metrics

KafkaRequestHandlerPool is a KafkaMetricsGroup with the following performance metrics.

Table 1. KafkaRequestHandlerPool’s Performance Metrics
Metric Name	Description
`requestHandlerAvgIdleMetricName`	Aggregare idle meter - average free capacity of the request handlers (idle handlers)

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

Figure 2. KafkaMetricsGroup in jconsole

Creating and Starting KafkaRequestHandler Daemon Thread — `createHandler` Method

createHandler(id: Int): Unit

createHandler creates a new KafkaRequestHandler for the given id (and the broker ID, the aggregare idle meter, the threadPoolSize, the RequestChannel and the KafkaApis).

createHandler adds the KafkaRequestHandler to the runnables internal registry.

In the end, createHandler starts the KafkaRequestHandler as a daemon thread with the name with the format:

[logAndThreadNamePrefix]-kafka-request-handler-[id]

Note	`createHandler` is used when `KafkaRequestHandlerPool` is created (to create the numThreads handlers) and requested to resize the thread pool (with new handlers).

Resizing Thread Pool (of KafkaRequestHandlers) — `resizeThreadPool` Method

resizeThreadPool(newSize: Int): Unit

resizeThreadPool prints out the following INFO message to the logs:

Resizing request handler thread pool size from [currentSize] to [newSize]

When the given newSize is greater than the current threadPoolSize, resizeThreadPool creates and starts new KafkaRequestHandler daemon threads.

When the given newSize is smaller than the current threadPoolSize, resizeThreadPool removes KafkaRequestHandlers from the runnables internal registry and requests them to stop.

In the end, resizeThreadPool sets the threadPoolSize internal registry to the given newSize.

Note	`resizeThreadPool` is used exclusively when `DynamicThreadPool` is requested to reconfigure (the num.io.threads configuration property).

Shutting Down — `shutdown` Method

shutdown(): Unit

shutdown…FIXME

Note	`shutdown` is used when…FIXME

Internal Properties

Name Description

Name	Description
`runnables`	Pool of KafkaRequestHandlers daemon threads
`threadPoolSize`

runnables

Pool of KafkaRequestHandlers daemon threads

threadPoolSize

KafkaRequestHandlerPool