Digging into Kafka's Source 4: The Broker's Network Layer — Acceptors, Processors and Request Handlers

What this post covers

This is the fourth post in the "Digging into Kafka's Source" series. In part 2 we drew the big map of brokers, controllers and partitions, and in part 3 we looked at the shape of the bytes that clients and brokers exchange (a 4-byte length + a header + a body). This time we follow those bytes after they arrive at the broker: accepting the connection, reading the request, handing it to a thread that processes it, and writing the response back to the socket.

The blue and yellow boxes are the broker's network layer, the subject of this post. The grey boxes belong to the posts before and after it.

In terms of the part 2 map, this post zooms in on all of region B (the broker's network layer) and the entrance to region C (request handling) — the request handler threads and the branches of KafkaApis.handle().

There are four questions to answer.

• What do the data-plane-kafka-socket-acceptor-…, …-network-thread-… and …-request-handler-N threads in a broker's jstack each do, and how many of each are there?
• Which thread reads a request's bytes, which one processes it, and who writes the response to the socket?
• With several processing threads, how is the order of the requests sent on one connection preserved?
• Where in the code are num.network.threads, num.io.threads and queued.max.requests used, and what stops when they overflow?

All line numbers in this post refer to this commit. Part 1 shows how to check out the same commit and build the Docker image used for the examples. The network layer we read here is almost identical to 4.4.0, the latest stable release most people run. Comparing SocketServer.scala, RequestChannel.scala, KafkaRequestHandler.scala, Selector.java, KafkaChannel.java and Request.java with the 4.4.0 tag, only a few lines in SocketServer.scala that read the quota settings (where a QuotaConfig is built) differ; the rest is the same.

SocketServer.scala alone is 1,695 lines, and besides the Acceptor and the Processor it also holds connection limits (ConnectionQuotas), connection-creation-rate quotas and code for adding and removing listeners at runtime. This post picks only the main line a newcomer can follow; whatever was left out is listed by name and location in "What this post leaves out" at the end.

Words to know first

A listener is one address on which the broker accepts connections. In server.properties you list them as name://host:port, separated by commas, as in listeners=PLAINTEXT://:9092,CONTROLLER://:9093. Each name gets a security protocol (PLAINTEXT, SSL, SASL_…) through listener.security.protocol.map. The examples in this post run the broker and the controller in one process (part 2's combined mode), so there are two listeners: PLAINTEXT (9092) for clients, and CONTROLLER (9093) for controller traffic.

NIO Selector: Java's tool for letting one thread watch many sockets. You register socket channels with an interest — "tell me when it becomes readable (OP_READ)", "tell me when a new connection arrives (OP_ACCEPT)" — and call select(); it waits until some channel is ready and returns the list. On Linux there is epoll underneath. It is the very same tool the Poller used in Tomcat part 5.

Selector (Kafka's) and KafkaChannel: Kafka doesn't use the JDK's Selector directly; it wraps it in a class called org.apache.kafka.common.network.Selector. This class does the work of "read the 4-byte length, then read that many more bytes to complete one request (or response)" and "keep writing the bytes to send, bit by bit, until they're all out". One connection is one KafkaChannel object, which holds one receive in progress (NetworkReceive) and one send in progress (NetworkSend). Both classes live in the clients module, so producers and consumers handle the network with the same classes as the broker (we meet them again in part 5). Because the names collide, this post calls the JDK's one the "NIO Selector" and Kafka's one "Selector", or KSelector as the source aliases it.

Connection id: the broker gives every connection a string id such as 127.0.0.1:9092-127.0.0.1:36712-1-1. It is localAddress:port-remoteAddress:port-processorNumber-index (ServerConnectionId.java lines 122–128). The trailing index keeps ids from colliding even when the same remote port is reused (the comment at SocketServer.scala lines 866–868). Thanks to the third field, the request log alone tells you which Processor handled a connection (Example 3 in section 6).

mute: putting a connection into a "don't read it for now" state. In the source it is implemented by removing OP_READ from that channel's interest set in the NIO Selector. If the client sends more bytes, they just pile up in the operating system's socket buffer; the broker doesn't read them. Section 6 of this post is about why mute is needed.

Request queue and response queues: Processor threads and request-handler threads don't call each other directly; they pass work through queues. Requests go into one queue shared by all Processors (RequestChannel.requestQueue); responses go into a separate queue per Processor (Processor.responseQueue).

1. Who builds the SocketServer, and how many threads appear

When a broker starts, BrokerServer.startup() calls new SocketServer(...) (BrokerServer.scala lines 289–295), and if the process also has the controller role, ControllerServer.startup() builds a second SocketServer of its own (ControllerServer.scala lines 182–187). We saw what BrokerServer and ControllerServer are in part 2. The constructor creates an Acceptor for each listener it is responsible for.

  // Create acceptors and processors for the statically configured endpoints when the
  // SocketServer is constructed. Note that this just opens the ports and creates the data
  // structures. It does not start the acceptors and processors or their associated JVM
  // threads.
  if (apiVersionManager.listenerType.equals(ListenerType.CONTROLLER)) {
    config.controllerListeners.foreach(createDataPlaneAcceptorAndProcessors)
  } else {
    config.dataPlaneListeners.foreach(createDataPlaneAcceptorAndProcessors)
  }

core/src/main/scala/kafka/network/SocketServer.scala lines 142–150

Reading the Scala — This block is a statement sitting directly in the class body, so it runs in the constructor (Scala has no separate constructor method; the class body is the constructor). xs.foreach(f) is the same as Java's xs.forEach(this::f) — you pass the method name as is.

The controller's SocketServer takes only the listeners named in controller.listener.names, and the broker's takes the rest (dataPlaneListeners is all listeners minus the controller listeners, KafkaConfig.scala lines 209–219). As the comment says, at this stage only data structures are created; no threads are started yet. The threads start later, when enableRequestProcessing() calls Acceptor.start() (lines 173–211). The idea is to start taking requests only after things like the credentials used for authentication are ready (the comment at BrokerServer.scala lines 286–288).

Processors: num.network.threads per listener

The DataPlaneAcceptor that createDataPlaneAcceptorAndProcessors() builds calls addProcessors(value of num.network.threads) as soon as it receives its configuration (lines 450–452). Here is addProcessors().

  def addProcessors(toCreate: Int): Unit = synchronized {
    val listenerName = ListenerName.normalised(endPoint.listener)
    val securityProtocol = endPoint.securityProtocol
    val listenerProcessors = new ArrayBuffer[Processor]()

    for (_ <- 0 until toCreate) {
      val processor = newProcessor(socketServer.nextProcessorId(), listenerName, securityProtocol, socketServer.connectionDisconnectListeners)
      listenerProcessors += processor
      requestChannel.addProcessor(processor)

      if (started.get) {
        processor.start()
      }
    }
    processors ++= listenerProcessors
  }

core/src/main/scala/kafka/network/SocketServer.scala lines 725–740

Reading the Scala — def f(x: Int): Unit = synchronized { … } is a method that returns nothing (Unit) whose whole body is a synchronized block (like a Java synchronized method). val is a variable you can't reassign (final), for (_ <- 0 until n) means "repeat n times", buf += x appends one element and buf ++= xs appends several.

Two things to notice:

1. Processor numbers are assigned per SocketServer (socketServer.nextProcessorId(), lines 154–156). The broker-side SocketServer's Processors are 0, 1, 2 …, and the controller-side SocketServer's Processors are separately 0, 1, 2 ….
2. Each new Processor is registered with requestChannel.addProcessor(). Later, when a request handler sends a response back, it uses the Processor number recorded in the request to look the Processor up in this table (section 5).

newProcessor() names the thread.

    val name = s"data-plane-kafka-network-thread-$nodeId-${endPoint.listener}-${endPoint.securityProtocol}-$id"

core/src/main/scala/kafka/network/SocketServer.scala line 746

s"…$x…${expr}…" is Scala's string interpolation (Java's "…" + x + "…"). The node id, listener name, security protocol and Processor number are appended in that order. There is one Acceptor thread per listener, named s"data-plane-kafka-socket-acceptor-${endPoint.listener}-${endPoint.securityProtocol}-${endPoint.port}" (lines 506–508). It ends with the port, not a Processor number.

Request handlers: num.io.threads per pool

The threads that actually process requests are created outside SocketServer. After building KafkaApis, BrokerServer calls createPool(..., config.numIoThreads, "broker") (BrokerServer.scala lines 518–525), and ControllerServer likewise passes ControllerApis to createPool(..., config.numIoThreads, "controller") (ControllerServer.scala lines 298–305). The pool starts its threads like this.

  private def createHandler(id: Int): Unit = {
    runnables += new KafkaRequestHandler(
      id,
      brokerId,
      aggregateIdleMeter,
      aggregateThreads,
      perPoolIdleMeter,
      threadPoolSize,
      requestChannel,
      apis,
      time,
      nodeName
    )
    aggregateThreads.getAndIncrement()
    KafkaThread.daemon("data-plane-kafka-request-handler-" + id, runnables(id)).start()
  }

core/src/main/scala/kafka/server/KafkaRequestHandler.scala lines 261–276

The nodeName ("broker"/"controller") does not go into the name. So when you run the broker and the controller in one process, you get two sets of threads with the same names. You'll see it in the example below.

Example 1: the threads of a server started with default settings

This example was run with Kafka 4.5.0-SNAPSHOT built as-is from the reference commit (ab1e816411), in a Docker container (Eclipse Temurin JDK 25). The example folder is ex/u04-socket-server; its run.sh starts a broker+controller process with default settings, picks out the network-layer thread names from jstack, replaces the trailing number with N and counts them. The output in this section and in sections 3, 6 and 7 all comes from a single run of the same run.sh.

# Extract only the thread names, turn the trailing number into N and count (network-layer threads only)
thread_groups() {
  jstack "$1" | grep '^"' | sed 's/^"\([^"]*\)".*/\1/' \
 grep -E 'acceptor|network-thread|request-handler' | sed -E 's/[0-9]+$/N/' | sort | uniq -c
}

[From the example's ex/u04-socket-server/run.sh]

=== 1. default settings: network-layer threads (trailing number -> N)
process.roles=broker,controller
listeners=PLAINTEXT://:9092,CONTROLLER://:9093
      3 data-plane-kafka-network-thread-1-CONTROLLER-PLAINTEXT-N
      3 data-plane-kafka-network-thread-1-PLAINTEXT-PLAINTEXT-N
     16 data-plane-kafka-request-handler-N
      1 data-plane-kafka-socket-acceptor-CONTROLLER-PLAINTEXT-N
      1 data-plane-kafka-socket-acceptor-PLAINTEXT-PLAINTEXT-N
--- full names of the network threads
"data-plane-kafka-network-thread-1-CONTROLLER-PLAINTEXT-0"
"data-plane-kafka-network-thread-1-CONTROLLER-PLAINTEXT-1"
"data-plane-kafka-network-thread-1-CONTROLLER-PLAINTEXT-2"
"data-plane-kafka-network-thread-1-PLAINTEXT-PLAINTEXT-0"
"data-plane-kafka-network-thread-1-PLAINTEXT-PLAINTEXT-1"
"data-plane-kafka-network-thread-1-PLAINTEXT-PLAINTEXT-2"

Four things to read here:

• One Acceptor per listener, so two. The N at the end replaced the port (9092, 9093).
• Three Processors per listener, because the default of num.network.threads is 3 (section 7). In the name, 1 is the node id and PLAINTEXT-PLAINTEXT means "listener name PLAINTEXT, security protocol PLAINTEXT". The controller listener is named CONTROLLER, but its security protocol is also PLAINTEXT.
• Numbering restarts at 0 for each listener, because the two listeners belong to different SocketServers. If a broker had two listeners of its own (say, internal and external), they'd be in the same SocketServer and numbered 0–5 in a row.
• There are 16 request-handlers. A pool of 8 (the default of num.io.threads) exists once for the broker and once for the controller, both named …-handler-0 to …-handler-7.

Blue boxes are threads (the bottom sendResponse() is a step the request handler thread calls); the yellow box is a queue shared by several threads. The picture shows only one SocketServer and one listener.

Easy to get wrong

• "data-plane" is a prefix that means almost nothing today. In the ZooKeeper era you could have a separate listener just for the requests the controller sent to brokers (control.plane.listener.name), and the other side was called the "data plane". That setting was removed in commit 0c435e3855 (2025-01-08, "KAFKA-18353 Remove zk config control.plane.listener.name"), and this commit has no control-plane code. Even KRaft's controller listener (CONTROLLER) has threads named data-plane-….
• The setting's description and reality differ. The documentation string of num.network.threads says "each listener (except for controller listener) creates its own thread pool" (SocketServerConfigs.java line 153), but as Example 1 shows, this commit creates the same number of Processors for the CONTROLLER listener too. The controller-side SocketServer goes down the same createDataPlaneAcceptorAndProcessors() → configure() → addProcessors() path.
• There are Java classes with the same names. SocketServer.java (42 lines) and Processor.java (43 lines) under server/src/main/java/org/apache/kafka/network/ are small classes holding only constants and helper methods; the thread bodies are the Scala kafka.network.SocketServer and kafka.network.Processor. The Scala side tells them apart with aliases: import org.apache.kafka.network.{Processor => JProcessor, SocketServer => JSocketServer, …} (line 43). When you see JProcessor.CONNECTION_QUEUE_SIZE in the source, it's the Java-side constant.

2. The Acceptor: accept connections and hand them to Processors in turn

The Acceptor's run() is simple. It registers the listening socket with its own NIO Selector for OP_ACCEPT, and repeats two jobs until told to stop.

  override def run(): Unit = {
    serverChannel.register(nioSelector, SelectionKey.OP_ACCEPT)
    try {
      while (shouldRun.get()) {
        try {
          acceptNewConnections()
          closeThrottledConnections()
        }
        catch {
          // We catch all the throwables to prevent the acceptor thread from exiting on exceptions due
          // to a select operation on a specific channel or a bad request. We don't want
          // the broker to stop responding to requests from other clients in these scenarios.
          case e: ControlThrowable => throw e
          case e: Throwable => error("Error occurred", e)
        }
      }
    } finally {
      closeAll()
    }
  }

core/src/main/scala/kafka/network/SocketServer.scala lines 572–591

acceptNewConnections() is the core.

  private def acceptNewConnections(): Unit = {
    val ready = nioSelector.select(500)
    if (ready > 0) {
      val keys = nioSelector.selectedKeys()
      val iter = keys.iterator()
      while (iter.hasNext && shouldRun.get()) {
        try {
          val key = iter.next
          iter.remove()

          if (key.isAcceptable) {
            accept(key).foreach { socketChannel =>
              // Assign the channel to the next processor (using round-robin) to which the
              // channel can be added without blocking. If newConnections queue is full on
              // all processors, block until the last one is able to accept a connection.
              var retriesLeft = synchronized(processors.length)
              var processor: Processor = null
              do {
                retriesLeft -= 1
                processor = synchronized {
                  // adjust the index (if necessary) and retrieve the processor atomically for
                  // correct behaviour in case the number of processors is reduced dynamically
                  currentProcessorIndex = currentProcessorIndex % processors.length
                  processors(currentProcessorIndex)
                }
                currentProcessorIndex += 1
              } while (!assignNewConnection(socketChannel, processor, retriesLeft == 0))
            }
          } else
            throw new IllegalStateException("Unrecognized key state for acceptor thread.")
        } catch {
          case e: Throwable => error("Error while accepting connection", e)
        }
      }
    }
  }

core/src/main/scala/kafka/network/SocketServer.scala lines 623–658

Reading the Scala — accept(key) returns an Option[SocketChannel]. An Option is a box that either holds a value (Some(x)) or doesn't (None), and opt.foreach { x => … } runs the block only when there is a value (Java's Optional.ifPresent). var is a variable you can reassign, and synchronized(expr) is the value of that expression computed under the lock.

Reading one turn in order:

1. nioSelector.select(500) — wait up to 500 ms for a new connection. It doesn't wait forever so that it can check shouldRun at the top of the loop and, below, run closeThrottledConnections() to clean up "connections that hit the connection-rate quota and were deliberately scheduled to be closed late".
2. accept(key) — take the connection with serverSocketChannel.accept(), and if it passes the connection limits (ConnectionQuotas.inc()), configure the socket and return Some(socketChannel) (lines 663–687). If it exceeds the per-IP limit (max.connections.per.ip), it is closed at once and the result is None.
3. Hand it to a Processor chosen round robin. It bumps currentProcessorIndex and calls accept() on the next Processor. If that Processor's waiting queue is full, it moves on to the next one, and only on the last attempt (retriesLeft == 0) does it wait until there is room (mayBlock).

The accepted socket is configured like this.

  protected def configureAcceptedSocketChannel(socketChannel: SocketChannel): Unit = {
    socketChannel.configureBlocking(false)
    socketChannel.socket().setTcpNoDelay(true)
    socketChannel.socket().setKeepAlive(true)
    if (sendBufferSize != Selectable.USE_DEFAULT_BUFFER_SIZE)
      socketChannel.socket().setSendBufferSize(sendBufferSize)
  }

core/src/main/scala/kafka/network/SocketServer.scala lines 689–695

It makes the socket non-blocking (from now on it's handled by a Selector) and turns on TCP_NODELAY. TCP_NODELAY disables the operating system's habit of waiting briefly to coalesce small packets (Nagle's algorithm). Kafka's batches are already gathered by the application, so there's no reason for the OS to wait once more.

The Processor-side accept() just puts the socket in a queue and wakes the Processor.

  def accept(socketChannel: SocketChannel,
             mayBlock: Boolean,
             acceptorBlockedPercentMeter: com.yammer.metrics.core.Meter): Boolean = {
    val accepted = {
      if (newConnections.offer(socketChannel))
        true
      else if (mayBlock) {
        val startNs = time.nanoseconds
        newConnections.put(socketChannel)
        acceptorBlockedPercentMeter.mark(time.nanoseconds() - startNs)
        true
      } else
        false
    }
    if (accepted)
      wakeup()
    accepted
  }

core/src/main/scala/kafka/network/SocketServer.scala lines 1119–1136

newConnections is an ArrayBlockingQueue of size 20 (line 809; the size is the Java-side constant CONNECTION_QUEUE_SIZE = 20, Processor.java line 31). offer() returns false immediately when there's no room; put() waits until there is. The Acceptor doesn't register the socket with the Processor's Selector itself; it only queues it. The registration is done by the Selector's owner, the Processor thread, in configureNewConnections() on its next turn (section 3). The source doesn't state the reason in a comment, but it has the same shape as Tomcat's Acceptor, which only put sockets in the Poller's queue — a structure in which only one thread touches an NIO Selector. wakeup() wakes a Processor that may be waiting in select() so it registers the new connection right away.

The Acceptor only puts sockets in a queue; registering with the Selector is done by the Processor that owns it. Three Processors is the picture with the default num.network.threads=3.

One more piece of Scala — as with the last line of this method, accepted, the last expression of a block is the block's value (its return value) in Scala. In val accepted = { … } too, the value of the if expression inside the braces becomes accepted.

Easy to get wrong: how it differs from Tomcat's Acceptor

The Acceptor in Tomcat part 5 kept its listening socket blocking and simply waited in accept(). Kafka's Acceptor registers even the listening socket with an NIO Selector and waits in select(500). That's why in jstack Kafka's Acceptor sits in EPoll.wait, not in accept (Example 2 in section 3). Also, Tomcat had one Acceptor and one Poller per connector, while Kafka has one Acceptor per listener and several Processors per listener.

Connection limits sit in a different place, too. In Tomcat, when maxConnections was exceeded, the Acceptor stopped in a LimitLatch before accept(). Kafka first takes the connection with accept() and then, inside ConnectionQuotas.inc(), waitForConnectionSlot() checks the broker-wide and per-listener limits (max.connections) and the connection-creation rate; if they're exceeded, the Acceptor stops in counts.wait() (lines 1431–1448). All the defaults are Integer.MAX_VALUE, so none of them bite unless you set them (the table in section 7). The equivalent of the kernel's accept queue size (Tomcat's acceptCount) is socket.listen.backlog.size (default 50), passed when the listening socket is bound with bind(socketAddress, listenBacklogSize) (ServerSocketFactory.java line 53).

3. The Processor: one thread does both the reading and the writing

A Processor thread's run() goes through seven steps per turn.

  override def run(): Unit = {
    try {
      while (shouldRun.get()) {
        try {
          // setup any new connections that have been queued up
          configureNewConnections()
          // register any new responses for writing
          processNewResponses()
          poll()
          processCompletedReceives()
          processCompletedSends()
          processDisconnected()
          closeExcessConnections()
        } catch {
          // ...

core/src/main/scala/kafka/network/SocketServer.scala lines 871–884

This is where it differs most from Tomcat. Tomcat's Poller only announced "there's something to read", and the worker thread did the reading. Kafka's Processor reads by itself until an entire request has arrived, and writes the resulting response by itself too. Request handlers never touch a socket. As we saw in part 3, every Kafka request carries its length in the first 4 bytes, so a network thread can tell "where one request ends" without understanding its contents — that's what makes this split possible. The code that reads the length and then that many bytes is NetworkReceive.readFrom(), and if the length is larger than socket.request.max.bytes (default 100MB) it throws InvalidReceiveException (NetworkReceive.java lines 94–95).

Reading and writing sockets is this one thread's job from start to finish; it is connected to the request handler threads only through two queues.

When a request is complete: build a Request, queue it, mute

  private def processCompletedReceives(): Unit = {
    selector.completedReceives.forEach { receive =>
      var header: RequestHeader = null
      var req: Request = null
      try {
        openOrClosingChannel(receive.source) match {
          case Some(channel) =>
            header = JProcessor.parseRequestHeader(apiVersionManager, receive.payload)
            // ...
              val nowNanos = time.nanoseconds()
              // ...
                val connectionId = receive.source
                val context = new RequestContext(header, connectionId, channel.socketAddress, Optional.of(channel.socketPort()),
                  channel.principal, listenerName, securityProtocol, channel.channelMetadataRegistry.clientInformation,
                  isPrivilegedListener, channel.principalSerde)

                req = new Request(id, context, nowNanos, memoryPool, receive.payload, requestChannel.metrics)
                // ...
                requestChannel.sendRequest(req)
                selector.mute(connectionId)
                handleChannelMuteEvent(connectionId, ChannelMuteEvent.REQUEST_RECEIVED)

core/src/main/scala/kafka/network/SocketServer.scala lines 982–1021

Reading the Scala — x match { case Some(channel) => … case None => … } is a broader form of Java's switch; it opens the Option box and, if there's a value, takes it out under the name channel. The elided parts contain SASL re-authentication, closing connections whose authentication session has expired, and recording the client's name and version from an ApiVersions request (KIP-511, lines 1009–1018).

1. Only the header is parsed first. parseRequestHeader() reads the header and checks that the API and version are usable on this listener. If not, it throws an exception (InvalidRequestException or UnsupportedVersionException, Processor.java lines 33–42), and processChannelException() in the catch closes the connection. That's why sending a request in a version the broker doesn't know gets the connection dropped without a response (only ApiVersions passes in any version — lines 247–250 of ApiKeys.isVersionEnabled(); the reason lies in the version negotiation of part 3).
2. A RequestContext and a Request are built. The context holds "who sent it, on which listener, with which security protocol"; the Request holds that context, the body, and also the arrival time nowNanos and the Processor number id. The arrival time is used for the time breakdown in the request log, and the Processor number to find where the response must go. The Request constructor parses the body as well (context.parseRequest(buffer), Request.java line 101), so turning bytes into objects is also the network thread's job.
3. It's put on the request queue (sendRequest) and the connection is muted. This means: don't read anything more from this connection until this request's response has gone out. Section 6 explains why.

Example 2: where each thread waits while idle

Before sending any requests, the same run.sh picked one thread of each kind and took a jstack. Only the top frame and the Kafka frames are kept (the thread_stack function in run.sh).

=== 2. where each kind of thread waits while idle
"data-plane-kafka-socket-acceptor-PLAINTEXT-PLAINTEXT-9092" #76 [1866] prio=5 os_prio=0 cpu=2.09ms elapsed=2.93s tid=0x00007efdbd3cbb00 nid=1866 runnable  [0x00007efc755d4000]
	at sun.nio.ch.EPoll.wait(java.base@25.0.5/Native Method)
	at kafka.network.Acceptor.acceptNewConnections(SocketServer.scala:624)
	at kafka.network.Acceptor.run(SocketServer.scala:577)
"data-plane-kafka-network-thread-1-PLAINTEXT-PLAINTEXT-0" #77 [1863] prio=5 os_prio=0 cpu=18.97ms elapsed=3.14s tid=0x00007efdbd3c30a0 nid=1863 runnable  [0x00007efc758d7000]
	at sun.nio.ch.EPoll.wait(java.base@25.0.5/Native Method)
	at org.apache.kafka.common.network.Selector.select(Selector.java:880)
	at org.apache.kafka.common.network.Selector.poll(Selector.java:470)
	at kafka.network.Processor.poll(SocketServer.scala:973)
	at kafka.network.Processor.run(SocketServer.scala:879)
"data-plane-kafka-request-handler-0" #59 [1544] daemon prio=5 os_prio=0 cpu=13.59ms elapsed=4.23s tid=0x00007efdbd2085d0 nid=1544 waiting on condition  [0x00007efd29bee000]
	at jdk.internal.misc.Unsafe.park(java.base@25.0.5/Native Method)
	at kafka.network.RequestChannel.receiveRequest(RequestChannel.scala:165)
	at kafka.server.KafkaRequestHandler.run(KafkaRequestHandler.scala:117)

Values such as #76, nid, tid, cpu and elapsed differ on every run. All line numbers were checked against the reference commit's source.

• The Acceptor is in EPoll.wait at SocketServer.scala:624, which is nioSelector.select(500). It's waiting for a new connection.
• The Processor is also in EPoll.wait, via poll() in Processor.run (line 879) → selector.poll(pollTimeout) in Processor.poll (line 973) → Selector.poll in the clients module → Selector.select (line 880, nioSelector.select(timeoutMs)). The stack shows plainly that the broker's network thread waits inside the client library's org.apache.kafka.common.network.Selector.
• The request handler waits at RequestChannel.scala:165, which is requestQueue.poll(timeout, …), for work to arrive in the queue (the JDK's ArrayBlockingQueue frames were filtered out).
• Only the request handler's header line says daemon. The Acceptor and Processor are created with KafkaThread.nonDaemon(…) (lines 506 and 807), and request handlers with KafkaThread.daemon(…) (KafkaRequestHandler.scala line 275).

4. The RequestChannel and the request handlers

The numbers are the order referred to in the text. Steps 1–7 are sections 2 and 3, 8–9 this section, 10–13 sections 5 and 6.

The request queue: one, shared by all Processors

There is one RequestChannel per SocketServer (line 100), and the request queue is a single ArrayBlockingQueue inside it.

  private val requestQueue = new ArrayBlockingQueue[BaseRequest](queueSize)

core/src/main/scala/kafka/network/RequestChannel.scala line 50

queueSize is the config.queuedMaxRequests passed in by SocketServer — that is, queued.max.requests (default 500) (SocketServer.scala lines 85 and 100). Here are the putting side and the taking side.

  /** Send a request to be handled, potentially blocking until there is room in the queue for the request */
  def sendRequest(request: Request): Unit = {
    requestQueue.put(request)
  }

core/src/main/scala/kafka/network/RequestChannel.scala lines 76–79

  def receiveRequest(timeout: Long): BaseRequest = {
    val callbackRequest = callbackQueue.poll()
    if (callbackRequest != null)
      callbackRequest
    else {
      val request = requestQueue.poll(timeout, TimeUnit.MILLISECONDS)
      request match {
        case _: WakeupRequest => callbackQueue.poll()
        case _ => request
      }
    }
  }

core/src/main/scala/kafka/network/RequestChannel.scala lines 160–171

sendRequest() is a put(). When the queue is full with 500 requests, the Processor thread stops right here. Since one Processor looks after many connections (possibly hundreds), that means all of its connections pause reading and writing for a while. It's a back-pressure device: when the handlers can't keep up, the broker stops taking more from the network. The setting's description says the same: "the number of queued requests allowed for data-plane, before blocking the network threads" (SocketServerConfigs.java line 145).

receiveRequest() shows one more queue. callbackQueue holds the follow-up work (callbacks) of asynchronous tasks that finished on other threads, so that a request handler can continue them; since those have already waited in line once, they're taken first (the comment at lines 157–159). WakeupRequest is an empty marker put on the request queue to wake a thread waiting there when a callback arrives (lines 195–199). For this post it's enough to know the names.

One turn of a request handler

  def run(): Unit = {
    threadRequestChannel.set(requestChannel)
    while (!stopped) {
      // ...
      val req = requestChannel.receiveRequest(300)
      // ...
      req match {
        case _: ShutdownRequest =>
          debug(s"Kafka request handler $id on broker $brokerId received shut down command")
          completeShutdown()
          return

        // ...
        case request: Request =>
          try {
            request.requestDequeueTimeNanos(endTime)
            trace(s"Kafka request handler $id on broker $brokerId handling request $request")
            threadCurrentRequest.set(request)
            apis.handle(request, requestLocal)
          } catch {
            case e: FatalExitError =>
              completeShutdown()
              Exit.exit(e.statusCode)
            case e: Throwable => error("Exception when handling request", e)
          } finally {
            threadCurrentRequest.remove()
            request.releaseBuffer()
          }
        // ...
        case null => // continue
      }
    }

core/src/main/scala/kafka/server/KafkaRequestHandler.scala lines 108–184

Reading the Scala — case _: ShutdownRequest => means "if it is of that type (we don't use the value)", and case request: Request => means "if it is of that type, take it out as request" (like Java 21's switch pattern case Request request ->). case null => is when nothing arrived for 300 ms and poll() returned null.

It does only three things. It records in the request the time it was taken off the queue (requestDequeueTimeNanos — this fixes the "time spent waiting in the queue" in the request log), calls apis.handle(), and when done, releases the request body's buffer. apis is KafkaApis for the broker's pool and ControllerApis for the controller's. Both implement a single trait (Java's interface) called ApiRequestHandler (lines 37–40). ShutdownRequest is a "stop now" marker put on the queue once per thread when the pool shuts down (initiateShutdown(), line 198).

KafkaApis.handle(): branching on the apiKey

  override def handle(request: Request, requestLocal: RequestLocal): Unit = {
    // ...
      request.header.apiKey match {
        case ApiKeys.PRODUCE => handleProduceRequest(request, requestLocal)
        case ApiKeys.FETCH => handleFetchRequest(request)
        case ApiKeys.LIST_OFFSETS => handleListOffsetRequest(request)
        case ApiKeys.METADATA => handleTopicMetadataRequest(request)
        // ...
        case ApiKeys.CREATE_TOPICS => forwardToController(request)
        // ...
        case _ => throw new IllegalStateException(s"No handler for request api key ${request.header.apiKey}")
      }

core/src/main/scala/kafka/server/KafkaApis.scala lines 157–257

Reading the Scala — Here case ApiKeys.PRODUCE => is a pattern that checks equality with a value (a Java enum constant), not a type. case _ => is the default of a Java switch.

It's one big match on the apiKey in the request header we saw in part 3. At this commit there are 79 case ApiKeys.… branches, and 16 of them are not handled by the broker itself but handed to the controller with forwardToController() (creating and deleting topics, ACLs, partition reassignment, client quotas, features and so on — part 10). This match is the boundary between this post and the ones after it. We follow the PRODUCE branch in part 6 and FETCH in part 8.

In handle()'s finally, replicaManager.tryCompleteActions() makes another attempt at deferred work, and the time the work on the handler thread finished (apiLocalCompleteTimeNanos) is recorded (lines 261–270).

Easy to get wrong: a request handler is not held "until the response goes out"

A Produce with acks=all, or a Fetch waiting for data to accumulate, can take hundreds of milliseconds to seconds before its response goes out. The request handler isn't waiting all that time. Such requests are parked inside handle() as delayed operations ("send the response once the condition is met" — the purgatory, parts 8 and 9), and handle() returns at once. Later, another thread calls requestChannel.sendResponse() to send the response. In Example 3 of section 6, most of the Fetch's time shows up as remoteTimeMs (time spent waiting outside the handler thread); that's the trace of it.

5. The way back for the response

When KafkaApis has built a response object, it calls requestChannel.sendResponse(request, response). This method wraps it in a SendResponse prepared for turning into bytes (request.buildResponseSend(response)) (lines 91–101), records the times, and ends like this.

    val processor = processors.get(response.request.processor)
    // The processor may be null if it was shutdown. In this case, the connections
    // are closed, so the response is dropped.
    if (processor != null) {
      processor.enqueueResponse(response)
    }

core/src/main/scala/kafka/network/RequestChannel.scala lines 149–154

It finds the Processor by the number recorded in the request and puts the response in that Processor's response queue. A response always goes back to the Processor that read the request. Naturally so: the connection (socket) is registered only with that Processor's Selector. enqueueResponse() does responseQueue.put(response) and then wakeup() to rouse the Processor waiting in poll() (lines 1184–1187). The response queue is a LinkedBlockingDeque with no size limit (line 811).

On its next turn, in processNewResponses(), the Processor takes the responses out and handles them by kind.

  private def processNewResponses(): Unit = {
    var currentResponse: Response = null
    while ({currentResponse = dequeueResponse(); currentResponse != null}) {
      val channelId = currentResponse.request.context.connectionId
      try {
        currentResponse match {
          case response: NoOpResponse =>
            // There is no response to send to the client, we need to read more pipelined requests
            // that are sitting in the server's socket buffer
            updateRequestMetrics(response)
            trace(s"Socket server received empty response to send, registering for read: $response")
            // Try unmuting the channel. If there was no quota violation and the channel has not been throttled,
            // it will be unmuted immediately. If the channel has been throttled, it will be unmuted only if the
            // throttling delay has already passed by now.
            handleChannelMuteEvent(channelId, ChannelMuteEvent.RESPONSE_SENT)
            tryUnmuteChannel(channelId)

          case response: SendResponse =>
            sendResponse(response, response.responseSend)
          case response: CloseConnectionResponse =>
            updateRequestMetrics(response)
            trace("Closing socket connection actively according to the response code.")
            close(channelId)
          // ...
        }

core/src/main/scala/kafka/network/SocketServer.scala lines 915–945

The meaning of acks=0 becomes visible in the source here. The producer doesn't wait for a response and the broker doesn't write one. Instead, if something fails, dropping the connection is the only signal. The comment at KafkaApis.scala lines 513–515 explains exactly that.

Once a SendResponse has been fully written, poll() puts it in completedSends, and processCompletedSends() finishes the job.

  private def processCompletedSends(): Unit = {
    selector.completedSends.forEach { send =>
      try {
        val response = inflightResponses.remove(send.destinationId).getOrElse {
          throw new IllegalStateException(s"Send for ${send.destinationId} completed, but not in `inflightResponses`")
        }
        updateRequestMetrics(response)

        // Try unmuting the channel. If there was no quota violation and the channel has not been throttled,
        // it will be unmuted immediately. If the channel has been throttled, it will unmuted only if the throttling
        // delay has already passed by now.
        handleChannelMuteEvent(send.destinationId, ChannelMuteEvent.RESPONSE_SENT)
        tryUnmuteChannel(send.destinationId)

core/src/main/scala/kafka/network/SocketServer.scala lines 1041–1053

updateRequestMetrics() finalizes the request's timings (the request log line is also written here, section 6), and only after the response has been fully written does it unmute. Now the next request on this connection can be read.

A response goes back only to the Processor that read its request, and every branch ends by unmuting (or closing) the connection so the next request can be read.

6. One request at a time per connection — mute and unmute

There are 8 request handlers and one request queue. If requests A and B arrived back to back on one connection and both went on the queue, handler-3 could take A while handler-5 takes B at the same time. Then B's response might go out before A's, and for Produce requests to the same partition, the write order could even flip. The Kafka protocol promises that responses on one connection come back in the order the requests were sent (the client's NetworkClient relies on that promise to pair responses with requests — the correlation id of part 3), so that must not happen.

Kafka's answer is simple: take out only one request per connection at a time. Until that request's response has fully gone out, the connection isn't read. The description of Selector.poll() states the reason directly.

     * At most one entry is added to "completedReceives" for a channel in each poll. This is necessary to guarantee that
     * requests from a channel are processed on the broker in the order they are sent. Since outstanding requests added
     * by SocketServer to the request queue may be processed by different request handler threads, requests on each
     * channel must be processed one-at-a-time to guarantee ordering.

clients/src/main/java/org/apache/kafka/common/network/Selector.java lines 434–437

The mechanism has two layers.

1. At most one completed request per connection per poll(). pollSelectionKeys() doesn't read a connection that already has a completed request or that is muted.

                //if channel is ready and has bytes to read from socket or buffer, and has no
                //previous completed receive then read from it
                if (channel.ready() && (key.isReadable() || channel.hasBytesBuffered()) && !hasCompletedReceive(channel)
                        && !explicitlyMutedChannels.contains(channel)) {
                    attemptRead(channel);
                }

clients/src/main/java/org/apache/kafka/common/network/Selector.java lines 578–583

2. The connection is muted the moment its request goes on the queue. In section 3's processCompletedReceives(), the line right after requestChannel.sendRequest(req) was selector.mute(connectionId). Muting comes down to removing OP_READ from the interest set in the NIO Selector.

    void mute() {
        if (muteState == ChannelMuteState.NOT_MUTED) {
            if (!disconnected) transportLayer.removeInterestOps(SelectionKey.OP_READ);
            muteState = ChannelMuteState.MUTED;
        }
    }

    /**
     * Unmute the channel. The channel can be unmuted only if it is in the MUTED state. For other muted states
     * (MUTED_AND_*), this is a no-op.
     *
     * @return Whether or not the channel is in the NOT_MUTED state after the call
     */
    boolean maybeUnmute() {
        if (muteState == ChannelMuteState.MUTED) {
            if (!disconnected) transportLayer.addInterestOps(SelectionKey.OP_READ);
            muteState = ChannelMuteState.NOT_MUTED;
        }
        return muteState == ChannelMuteState.NOT_MUTED;
    }

clients/src/main/java/org/apache/kafka/common/network/KafkaChannel.java lines 252–271

maybeUnmute() unmutes only when the state is exactly MUTED. There are five states so that "the response hasn't gone out yet" and "tied up by a quota" can be tracked separately. If the response is written but the connection is still throttled by a quota, it isn't unmuted; and if the throttling ends but the response hasn't gone out, it isn't unmuted either.

The green and yellow states are the path an ordinary request takes; the grey states are taken only when a quota kicks in. The transition rules are copied from the comment at KafkaChannel.java lines 101–107 and handleChannelMuteEvent() at lines 274–312.

An ordinary request goes around like this: NOT_MUTED → (request queued) mute() to MUTED → REQUEST_RECEIVED to MUTED_AND_RESPONSE_PENDING → (response fully written) RESPONSE_SENT to MUTED → maybeUnmute() to NOT_MUTED. If a transition not in the rules arrives, handleChannelMuteEvent() throws IllegalStateException (lines 309–311). Mute also happens because of memory. If you cap the total size of request buffers with queued.max.request.bytes and a buffer can't be obtained, KafkaChannel.read() mutes itself (lines 414–417). The default is -1, so this cap is off (MemoryPool.NONE, SocketServer.scala line 97).

Example 3: the second request on connection A waits until the first is done

We checked this with a raw socket. On connection A we send a Fetch that takes 2 seconds and a quick ApiVersions back to back without waiting for responses, and 0.1 s later we send one ApiVersions on connection B. The Fetch goes to an empty partition with MinBytes=1 and MaxWaitMs=2000; the broker waits up to 2 seconds for at least 1 byte to accumulate before answering (long polling — part 8). The request bytes are built the same way as part 3's RawWire.java.

// An example written for this post (not part of Kafka's source)
// On connection A, send a "long-waiting Fetch" and an ApiVersions back to back without waiting for responses;
// 0.1 s later send one ApiVersions on connection B. Measure when each response arrives.
// ... (imports omitted)
public class Pipeline {
    static long t0;

    public static void main(String[] args) throws Exception {
        try (Socket a = new Socket("localhost", 9092); Socket b = new Socket("localhost", 9092)) {
            // ...
            // MinBytes=1, MaxWaitMs=2000 on an empty partition — with no data, the broker waits up to 2 s before answering
            FetchRequestData fetch = new FetchRequestData()
                .setReplicaId(-1).setMaxWaitMs(2000).setMinBytes(1)
                .setTopics(List.of(new FetchRequestData.FetchTopic().setTopic("u04")
                    .setPartitions(List.of(new FetchRequestData.FetchPartition()
                        .setPartition(0).setFetchOffset(0).setPartitionMaxBytes(1024)))));
            // ...
            // Connection A: write both requests to the socket at once (pipelining)
            a.getOutputStream().write(frame("u04-a", 1, (short) 12, fetch));
            a.getOutputStream().write(frame("u04-a", 2, (short) 4, apiVersions));
            log("conn A: sent FETCH(corr=1, maxWait=2000ms) and API_VERSIONS(corr=2) back to back");

            Thread.sleep(100);
            b.getOutputStream().write(frame("u04-b", 3, (short) 4, apiVersions));
            log("conn B: sent API_VERSIONS(corr=3)");

            log("conn B: response corr=" + readCorrelationId(inB));
            log("conn A: response corr=" + readCorrelationId(inA));
            log("conn A: response corr=" + readCorrelationId(inA));
        }
    }
    // ... frame(): [4-byte length][request header][request body], readCorrelationId(): the int32 at the front of the response header
}

[Shortened from the example's ex/u04-socket-server/Pipeline.java]

The Fetch is sent as v12, the last version that names topics by name (from v13 on, topic ids are used; FetchRequest.json). Before running this, run.sh turned the request logger (kafka.request.logger) up to DEBUG: it copies the distribution's config/log4j2.yaml, changes that logger's level: WARN to DEBUG, and passes it with KAFKA_LOG4J_OPTS="-Dlog4j2.configurationFile=/tmp/log4j2.yaml" (bin/kafka-server-start.sh, which starts the server, uses its default configuration file config/log4j2.yaml only when this variable is empty, lines 24–25). This logger writes one line of JSON per request to kafka-request.log, and run.sh pulls out just the fields we need.

=== 3. one request at a time per connection
 0.25s conn A: sent FETCH(corr=1, maxWait=2000ms) and API_VERSIONS(corr=2) back to back
 0.37s conn B: sent API_VERSIONS(corr=3)
 0.38s conn B: response corr=3
 2.31s conn A: response corr=1
 2.31s conn A: response corr=2
--- kafka-request.log for these three requests (fields shortened)
u04-b API_VERSIONS corr=3 conn=127.0.0.1:9092-127.0.0.1:36718-2-1 total=2.939 queue=0.862 local=1.433 remote=0.0 respQueue=0.176 send=0.466
u04-a FETCH corr=1 conn=127.0.0.1:9092-127.0.0.1:36712-1-1 total=2058.32 queue=3.027 local=38.954 remote=2015.709 respQueue=0.129 send=0.499
u04-a API_VERSIONS corr=2 conn=127.0.0.1:9092-127.0.0.1:36712-1-1 total=1.72 queue=0.322 local=0.956 remote=0.0 respQueue=0.102 send=0.339

Times, ports and millisecond values differ on every run (we ran it twice and confirmed the order and magnitudes are the same). The 0.25 s at the top is the time the example program spent loading the serialization classes for the first time.

1. Connection B's ApiVersions got its answer in 0.01 s. The broker as a whole wasn't stalled by the Fetch.
2. Connection A's ApiVersions (corr=2) arrived at 2.31 s, the same moment as the Fetch (corr=1). The request itself takes 1.72 ms, but it wasn't processed while the Fetch ahead of it waited 2 seconds, because connection A was muted.
3. The third field of conn in the request log shows Processor 1 handled connection A and Processor 2 handled connection B — the Acceptor dealt them out in turn (which number it starts from is set by how many connections reached this listener before — the earlier kafka-topics.sh and so on; in three runs during review it was 1 and 2 every time).

The fields of the request log are computed by Request.updateRequestMetrics() (Request.java lines 329–341). Matched to the moments we've seen in this post:

Here is what one log line originally looks like (the response body is shortened to {...}).

--- one raw line (the API_VERSIONS from conn B)
[2026-11-07 03:57:04,426] DEBUG Completed request:{"isForwarded":false,
  "requestHeader":{"requestApiKey":18,"requestApiVersion":4,"correlationId":3,"clientId":"u04-b","requestApiKeyName":"API_VERSIONS"},"request":{"clientSoftwareName":"u04","clientSoftwareVersion":"1.0"},"response":{...},"connection":"127.0.0.1:9092-127.0.0.1:36718-2-1","totalTimeMs":2.939,"requestQueueTimeMs":0.862,"localTimeMs":1.433,"remoteTimeMs":0.0,"throttleTimeMs":0,"responseQueueTimeMs":0.176,"sendTimeMs":0.466,"securityProtocol":"PLAINTEXT","principal":"User:ANONYMOUS","listener":"PLAINTEXT","clientInformation":{"softwareName":"unknown","softwareVersion":"unknown"}} (kafka.request.logger)

Easy to get wrong

• Time spent muted doesn't show up in the request log. From the client's side, connection A's ApiVersions waited about 2 seconds, but its totalTimeMs in the request log is 1.72 ms. The clock starts at "the time the Processor finished reading the request", and while the connection was muted that request hadn't been read yet — it was sitting in the operating system's socket buffer. When broker metrics look fast but the client sees slowness, it's a place to suspect that an earlier request on the same connection took a long time.
• The client's max.in.flight.requests.per.connection doesn't change this rule. The producer's default is 5 (the define at ProducerConfig.java lines 520–525). The producer may send up to five requests on one connection without waiting for responses, but the broker still processes them one at a time. Sending ahead only saves round trips — the next request is already in the broker's socket buffer, so it can be read the instant the connection is unmuted. To process more in parallel you need more connections (brokers and partitions); requests within one connection are always processed in turn. The producer side of the story continues in part 5.
• acks=0 is one at a time too. The broker just doesn't write a response; the request handler still has to finish and send a NoOpResponse before the connection is unmuted (the table in section 5). Having no response doesn't mean the broker slurps requests in as fast as they come.

7. The settings at a glance, and where things get stuck

Here are the settings that came up, as name → constant → default → where it's used. Every default was checked against the constant passed to define(…).

SocketServerConfigs.java lines 95–153, ServerConfigs.java lines 59–61

The distribution's config/server.properties also contains num.network.threads=3 and num.io.threads=8. These equal the defaults, so deleting them changes nothing.

Yellow boxes are waiting lines; red boxes say what stops at that spot. Grey is a device that is off by default.

Read the figure from the bottom up and you get what jams, in order, when the request handlers slow down. When the request queue fills, Processors stop; when a Processor stops, its newConnections fills; when all of those are full, the Acceptor stops; and when the Acceptor stops, the kernel's accept queue fills. Every queue has a fixed size, so the broker's memory doesn't grow without bound. Only the response queue is unbounded (section 5), but since there's one response per request and the request side is already throttled, it can't pile up forever (this is read from the structure, not from a comment in the source).

Example 4: changing the thread counts

At the end of the same run.sh, the server is wiped and restarted with num.network.threads=1 and num.io.threads=2.

=== 4. num.network.threads=1 num.io.threads=2
      1 data-plane-kafka-network-thread-1-CONTROLLER-PLAINTEXT-N
      1 data-plane-kafka-network-thread-1-PLAINTEXT-PLAINTEXT-N
      4 data-plane-kafka-request-handler-N
      1 data-plane-kafka-socket-acceptor-CONTROLLER-PLAINTEXT-N
      1 data-plane-kafka-socket-acceptor-PLAINTEXT-PLAINTEXT-N
stopped

One Processor per listener, and four request handlers in two pools of two. The number of Acceptors is the number of listeners regardless of settings.

Both values can be changed without stopping the server. num.network.threads is registered as reconfigurable per listener (DataPlaneAcceptor.ListenerReconfigurableConfigs, line 357), and changing it makes reconfigure() add or remove Processors (lines 434–445). You can't, however, go below half or above double the current value in one step (validateReconfiguration(), lines 418–421). For num.io.threads, KafkaRequestHandlerPool.resizeThreadPool() adds or stops threads (lines 283–296). The examples in this post didn't go as far as changing them at runtime.

Side by side with Tomcat's NIO connector

The left side is this post; the right side is Tomcat part 5. The Kafka side shows only the broker-side SocketServer with default settings and one listener (in combined mode the controller side has a second, identical set).

What they share is the overall frame: "a few threads watch many connections with a Selector, and a separate thread pool does the real work". What differs is who does the socket I/O. With HTTP you have to interpret the headers to know where a request ends, so Tomcat left reading to the worker thread (the side that knows the protocol). With Kafka the length comes first, so the network thread can read to the end on its own, and the processing threads are cut off from sockets entirely. The price is one extra path back to the network thread to write the response (the response queue and wakeup()).

What this post leaves out

• Connection-count and connection-rate quota calculations (ConnectionQuotas, ListenerConnectionQuota, delayed close DelayedCloseSocket): from SocketServer.scala line 1252. Request quotas and throttling (StartThrottlingResponse) are in server/src/main/java/org/apache/kafka/server/quota/
• TLS and SASL (SslTransportLayer, SaslServerAuthenticator, KafkaChannel.prepare(), re-authentication maybeBeginServerReauthentication): clients/src/main/java/org/apache/kafka/common/network/, …/common/security/authenticator/
• Adding and removing listeners at runtime (addListeners/removeListeners): SocketServer.scala lines 281–311
• KafkaRequestHandler.wrapAsyncCallback() and CallbackRequest, which bring asynchronous work finished on other threads back to a request handler: KafkaRequestHandler.scala lines 65–85
• RequestConvertToJson, which turns requests into JSON (the body of the request log): server/src/main/java/org/apache/kafka/network/RequestConvertToJson.java
• Metrics (NetworkProcessorAvgIdlePercent, RequestHandlerAvgIdlePercent, RequestQueueSize and so on): SocketServer.scala lines 116–141, RequestChannel.scala lines 54–60, KafkaRequestHandler.scala lines 243–253
• The controller-side handler ControllerApis.handle(): from line 94 of core/src/main/scala/kafka/server/ControllerApis.scala (part 10)
• The KRaft network thread for replicating the metadata log (a separate Selector built by KafkaRaftManager): line 231 of core/src/main/scala/kafka/raft/KafkaRaftManager.scala

Summary

What's next

We've followed the broker up to the point where a request is handed to KafkaApis.handle(). Next we cross over to the side that sends those requests. When you call KafkaProducer.send(), the record doesn't go out immediately: it gathers in a batch in the RecordAccumulator, and then the Sender thread — which uses the same Selector we saw in this post — bundles batches into a Produce request and sends it. Which if uses linger.ms and batch.size, and how the client's max.in.flight.requests.per.connection meshes with this post's mute, are covered in Part 5: Inside the Producer — From send() to Batches and the Sender Thread.

The Tomcat series' Part 5: The Connector and the NIO Endpoint is an HTTP server's answer to the same question — "which thread takes a connection, and who reads the request?" Read side by side, the two posts show how the shape of a protocol (a length prefix, or headers you must interpret to find the end) changes how threads divide the work. A server that handles connections with a single event loop is covered in the nginx series.

• Part 2: The Big Picture — Clients, Brokers, Controllers and Partitions
• Part 3: The Wire Protocol — Requests and Responses Generated from JSON Schemas
• Part 6: How the Broker Accepts Records — The Record Batch Format and the Produce Path
• Part 8: The Read Path — Fetch Requests, Long Polling and Zero-Copy
• httpd Part 8: MPMs — How prefork, worker and event Share Out Connections