Understanding Kafka's SocketServer: Acceptor, Processor, and RequestChannel Architecture

Kafka's broker is divided into several layers—network, API, log storage, and replica replication. This article focuses on the network layer, specifically the SocketServer , which is an NIO server handling client connections.

The SocketServer thread model consists of a single Acceptor thread that accepts all new connections, a configurable number of Processor threads (each with its own selector) that read requests from connections, and a set of Handler threads that process those requests and write responses back to the client.

During broker startup, the startup() method creates Processor instances and an Acceptor, then starts the Acceptor thread:

def startup() {
    val brokerId = config.brokerId
    var processorBeginIndex = 0
    endpoints.values.foreach { endpoint =>
        val protocol = endpoint.protocolType
        val processorEndIndex = processorBeginIndex + numProcessorThreads
        for (i <- processorBeginIndex until processorEndIndex) {
            processors(i) = new Processor(i, time, maxRequestSize, requestChannel, connectionQuotas, connectionsMaxIdleMs, protocol, config.values, metrics)
        }
        val acceptor = new Acceptor(endpoint, sendBufferSize, recvBufferSize, brokerId, processors.slice(processorBeginIndex, processorEndIndex), connectionQuotas)
        acceptors.put(endpoint, acceptor)
        Utils.newThread("kafka-socket-acceptor-%s-%d".format(protocol.toString, endpoint.port), acceptor, false).start()
        acceptor.awaitStartup()
        processorBeginIndex = processorEndIndex
    }
}

The Acceptor registers a ServerSocketChannel for OP_ACCEPT events. When a key becomes acceptable, the Acceptor accepts the connection, obtains a SocketChannel, and hands it to a Processor using a round‑robin strategy.

def run() {
    serverChannel.register(nioSelector, SelectionKey.OP_ACCEPT)
    startupComplete()
    var currentProcessor = 0
    while (isRunning) {
        val ready = nioSelector.select(500)
        if (ready > 0) {
            val keys = nioSelector.selectedKeys()
            val iter = keys.iterator()
            while (iter.hasNext && isRunning) {
                val key = iter.next
                iter.remove()
                if (key.isAcceptable) accept(key, processors(currentProcessor))
                currentProcessor = (currentProcessor + 1) % processors.length
            }
        }
    }
}

Each Processor maintains a queue of newly accepted SocketChannel objects, registers them for OP_READ on its selector, and continuously polls for readable events. It reads client requests, creates a RequestChannel.Request, and sends it to the global RequestChannel.

def accept(socketChannel: SocketChannel) {
    newConnections.add(socketChannel)
    wakeup()
}

private def configureNewConnections() {
    while (!newConnections.isEmpty) {
        val channel = newConnections.poll()
        val connectionId = ConnectionId(...).toString
        selector.register(connectionId, channel)
    }
}

The RequestChannel is a shared component that holds a global request queue and a response queue per Processor. Processors place requests into the request queue, while Handlers (KafkaRequestHandler threads) consume them, process the business logic, and place responses into the appropriate Processor's response queue.

class RequestChannel(val numProcessors: Int, val queueSize: Int) extends KafkaMetricsGroup {
    private val requestQueue = new ArrayBlockingQueue[RequestChannel.Request](queueSize)
    private val responseQueues = new Array[BlockingQueue[RequestChannel.Response]](numProcessors)
    def sendRequest(request: RequestChannel.Request) { requestQueue.put(request) }
    def sendResponse(response: RequestChannel.Response) {
        responseQueues(response.processor).put(response)
        responseListeners.foreach(_(response.processor))
    }
    def receiveRequest(timeout: Long): RequestChannel.Request = requestQueue.poll(timeout, TimeUnit.MILLISECONDS)
    def receiveResponse(processor: Int): RequestChannel.Response = responseQueues(processor).poll()
}

Processors poll their response queues, register OP_WRITE for pending responses, and after the selector reports completed sends, they clean up the in‑flight response structures.

private def processNewResponses() {
    var curr = requestChannel.receiveResponse(id)
    while (curr != null) {
        curr.responseAction match {
            case RequestChannel.NoOpAction => selector.unmute(curr.request.connectionId)
            case RequestChannel.SendAction =>
                selector.send(curr.responseSend)
                inflightResponses += (curr.request.connectionId -> curr)
            case RequestChannel.CloseConnectionAction => close(selector, curr.request.connectionId)
        }
        curr = requestChannel.receiveResponse(id)
    }
}

The overall flow is:

NetworkClient → ClientRequest (Send) → KafkaChannel → SocketServer → Processor → RequestChannel → KafkaRequestHandler → KafkaApis → response back through the same path, using selectors for non‑blocking I/O.

This detailed walkthrough clarifies how Kafka achieves high‑throughput, low‑latency networking by separating connection acceptance, request reading, and response writing across dedicated threads and leveraging NIO selectors.