Digging into Kafka's Source 2: The Big Picture — Clients, Brokers, Controllers and Partitions

What this post covers

This is the second post in "Digging into Kafka's Source". In Part 1 we fetched the source at a fixed commit, built it, and unfolded a map of the folders and of the main road a single record travels. This post is the big-picture map you spread out before reading the code line by line. Every post from Part 3 to Part 12 picks one region of the single map in this post (the figure in section 1) and zooms in.

If you have used Kafka from an application, you have written bootstrap.servers=localhost:9092 and seen lines like Leader: 1 Replicas: 1,2,3 Isr: 1,2,3 from kafka-topics.sh --describe. Who picked that leader, and how does your producer know to go to it? Using one server setting (process.roles) as a bridge, this post answers these questions:

• When you run kafka-server-start.sh server.properties, which classes are created in which order, and what exactly are a "broker" and a "controller" in the source?
• What does one broker hold — the components and threads it creates at startup?
• Who decides topics, partitions, replicas, leaders and the ISR, and how does that decision reach every broker?
• Which server does a client request go to — how a Metadata response leads you to the leader
• Which modules (clients, core, storage, raft, metadata …) depend on which?

All line numbers in this post refer to that commit. How to fetch the same commit and build the Docker image for the examples is in Part 1. Paths start at the repository root (core/src/main/scala/kafka/server/BrokerServer.scala).

Since this post draws a map, it only follows the main roads. Each section points to the post that reads that component in detail ("we'll see it in Part N").

Terms to know first

A broker is a server that keeps topic data on disk and answers producers' writes (Produce) and consumers' reads (Fetch). It is what your application connects to via bootstrap.servers.

A controller is the server that keeps the cluster's "ledger". It decides and records which topics exist, how many partitions each has, which brokers hold each partition's copies, and who the leader is. It does not hold the data (the records).

KRaft is the name for how controllers keep that ledger together using a consensus algorithm called Raft (Kafka + Raft). The ledger piles up as records in a metadata log called __cluster_metadata. An external system called ZooKeeper used to hold this ledger, but from 4.x on there is only KRaft, and this commit's core/src/main contains no ZooKeeper code. We look at Raft itself in Part 10.

process.roles is a line in the server config file that says whether this process is a broker (broker), a controller (controller), or both (broker,controller). Running both is called combined mode. The distribution's config/server.properties uses combined mode (process.roles=broker,controller, line 19), and it also ships config/broker.properties and config/controller.properties for running them separately.

Partitions, replicas, leaders and followers: a topic is split into partitions, and each partition keeps copies (replicas) on several brokers. That number is the replication factor (RF). One replica is the leader, which takes writes and reads; the other followers fetch data from the leader and keep up.

The ISR (in-sync replicas) is the list of replicas that are keeping up with the leader. That is the Isr: 1,2,3 in --describe. What "keeping up" means and how the ISR shrinks and grows is covered in Part 9.

A listener is a named port where a server accepts connections. The examples' config separates the one for clients (PLAINTEXT) from the one for controllers (CONTROLLER): listeners=PLAINTEXT://:9092,CONTROLLER://:9093.

1. One map — the seven regions of this series

Let's unfold the map first. This figure is the map for the whole series. From Part 3 on, each post begins by saying something like "this post zooms into region C of the Part 2 map", using the region letters in this figure.

The series map: seven regions A–G and the post that zooms into each. Blue boxes are components that process things; yellow boxes are where data piles up.

Notice the two roads on the map. The data road (A → B → C → D → E) is the road records travel; it is exactly the "main road of a single record" from Part 1. The ledger road (C ⇢ G → D) is the road a decision like "create this topic" takes: it is made in the controller and comes back to the brokers through the metadata log. The rest of this post checks, in the source, the skeleton that holds up these two roads — who builds these components and where each one lives.

2. How one process comes up: kafka.Kafka → KafkaRaftServer

The starting point: main()

The last line of bin/kafka-server-start.sh is exec $base_dir/kafka-run-class.sh $EXTRA_ARGS kafka.Kafka "$@" (bin/kafka-server-start.sh line 44). In the end, the main of a Scala object called kafka.Kafka is called.

  private def buildServer(props: Properties): Server = {
    val config = KafkaConfig.fromProps(props, doLog = false)
    new KafkaRaftServer(
      config,
      Time.SYSTEM,
    )
  }

  def main(args: Array[String]): Unit = {
    try {
      val serverProps = getPropsFromArgs(args)
      val server = buildServer(serverProps)
      // ...
      try server.startup()
      // ...
      server.awaitShutdown()

core/src/main/scala/kafka/Kafka.scala lines 65–106

Reading Scala — object Kafka (line 29) is a singleton with exactly one instance, so read it like a bag of Java static methods. def main(args: Array[String]): Unit is a method that takes a String[] and returns nothing (void), and val is a variable you cannot reassign (Java's final). As in buildServer, the last expression of a braced block is its return value, so there is no return.

buildServer() always builds a single KafkaRaftServer. There is no if choosing a server type. In the past this is where the ZooKeeper-mode server class was chosen too, but this commit has only the KRaft server. The first cut registers a handler that logs when the process is terminated by a signal (lines 78–85) and installs a hook that calls server.shutdown() when a termination signal arrives (lines 87–96); the second cut is the catch block that exits the process if startup fails (lines 99–104).

Building by role: KafkaRaftServer

The doc comment of KafkaRaftServer describes its job in one sentence:

It is responsible for constructing the controller and/or broker based on the process.roles configuration and for managing their basic lifecycle (startup and shutdown).

(KafkaRaftServer.scala lines 42–48) The constructor body is exactly that job:

  private val sharedServer = new SharedServer(
    config,
    // ...
  )

  private val broker: Option[BrokerServer] = if (config.processRoles.contains(ProcessRole.BrokerRole)) {
    Some(new BrokerServer(sharedServer))
  } else {
    None
  }

  private val controller: Option[ControllerServer] = if (config.processRoles.contains(ProcessRole.ControllerRole)) {
    Some(new ControllerServer(
      sharedServer,
      KafkaRaftServer.configSchema,
      bootstrapMetadata,
    ))
  } else {
    None
  }

  override def startup(): Unit = {
    Mx4jLoader.maybeLoad()
    // Controller component must be started before the broker component so that
    // the controller endpoints are passed to the KRaft manager
    controller.foreach(_.startup())
    broker.foreach(_.startup())
    AppInfoParser.registerAppInfo(Server.MetricsPrefix, config.brokerId.toString, metrics, time.milliseconds())
    info(KafkaBroker.STARTED_MESSAGE)
  }

core/src/main/scala/kafka/server/KafkaRaftServer.scala lines 65–100

Reading Scala — Option[BrokerServer] is "a value that may or may not be there": Some(value) if it is, None if not (the same idea as Java's Optional). Scala's if … else … is an expression that yields a value, so you can assign it directly: val broker = if (…) Some(…) else None. controller.foreach(_.startup()) means "if there is a value, call its startup()", and _ is shorthand for the lambda's argument (the c in c -> c.startup()).

Three things stand out.

1. The broker and the controller are two objects inside the same process. Depending on what process.roles contains, BrokerServer and ControllerServer each become Some or None. The only allowed values are "broker" and "controller" (define(PROCESS_ROLES_CONFIG, LIST, …, ValidList.in(false, "broker", "controller"), …) in KRaftConfigs, raft/src/main/java/org/apache/kafka/raft/KRaftConfigs.java line 79).
2. There is always exactly one SharedServer, whatever the roles (below in this section).
3. The controller starts first, and the broker shuts down first. The comment above explains the start order, and the comment in shutdown() (lines 102–109) explains that in combined mode the broker is stopped first "since the controller may be needed for controlled shutdown".

How one process comes up. SharedServer and the three components under it (yellow) exist once per process, whatever its roles.

What the two share: SharedServer

Here is the doc comment of SharedServer:

The SharedServer manages the components which are shared between the BrokerServer and ControllerServer. These shared components include the Raft manager, snapshot generator, and metadata loader. A KRaft server running in combined mode as both a broker and a controller will still contain only a single SharedServer instance.

(SharedServer.scala lines 83–96) The broker and the controller share three things:

If there were two of these in combined mode, the same log would be read twice and the same directory written twice, so they are bundled into one. Starting and stopping work just as the comment says:

  def startForBroker(): Unit = synchronized {
    if (!isUsed()) {
      start(Endpoints.empty())
    }
    usedByBroker = true
  }
  // ...
  def stopForBroker(): Unit = synchronized {
    if (usedByBroker) {
      usedByBroker = false
      if (!isUsed()) stop()
    }
  }

core/src/main/scala/kafka/server/SharedServer.scala lines 144–182

Reading Scala — def f(): Unit = synchronized { … } locks this for the whole method body, just like a Java synchronized method.

Whoever calls first actually starts it (start()), and whoever stops last actually stops it. According to the comment, two booleans (usedByBroker, usedByController) are used instead of a reference count to make debugging easier (lines 94–95). The skeleton of start(), which does the real work, looks like this:

        val _raftManager = new KafkaRaftManager[ApiMessageAndVersion](
          clusterId,
          raftManagerConfig,
          metaPropsEnsemble.logDirProps.get(metaPropsEnsemble.metadataLogDir.get).directoryId.get,
          MetadataRecordSerde.INSTANCE,
          KafkaRaftServer.MetadataPartition,
          KafkaRaftServer.MetadataTopicId,
          time,
          metrics,
          externalKRaftMetrics,
          Some(s"kafka-${sharedServerConfig.nodeId}-raft"), // No dash expected at the end
          // ...
        raftManager = _raftManager
        _raftManager.startup()
        // ...
        loader = loaderBuilder.build()
        // ...
        _raftManager.client.register(loader)

core/src/main/scala/kafka/server/SharedServer.scala lines 298–363

Reading Scala — the square brackets in KafkaRaftManager[ApiMessageAndVersion] are a type argument, like Java's <ApiMessageAndVersion>. The s"…" in s"kafka-${sharedServerConfig.nodeId}-raft" is string interpolation: ${expression} is spliced into the string.

The thread-name prefix kafka-<node id>-raft passed to KafkaRaftManager shows up again in section 3's example as kafka-1-raft-io-thread (the -io-thread suffix is added by raft/src/main/java/org/apache/kafka/raft/KafkaRaftClientDriver.java line 55). The last line, register(loader), asks: "when the Raft client commits new records, tell the MetadataLoader." That one line is the KafkaRaftClient → MetadataLoader arrow inside region G of the map in section 1.

Easy to get wrong: BrokerServer and ControllerServer are not Servers

By name all three are "servers", but the type main() works with, trait Server (three methods: startup(), shutdown(), awaitShutdown(), Server.scala lines 27–31), is implemented only by KafkaRaftServer (class KafkaRaftServer(…) extends Server with Logging, lines 49–52). BrokerServer implements a different trait, KafkaBroker (class BrokerServer(…) extends KafkaBroker, BrokerServer.scala lines 83–85), and ControllerServer only mixes in the logging trait (extends Logging, ControllerServer.scala lines 70–74). KafkaRaftServer represents the process; the other two are parts inside it. (Read trait like a Java interface.)

3. What one broker holds: BrokerServer.startup()

The components it builds while starting

BrokerServer.startup() is a long method of about 470 lines (lines 201–671). What it does is simple — build components one by one, hand them to each other, and open the doors at the very end. Picking only the lines that build components, in order:

  override def startup(): Unit = {
    if (!maybeChangeStatus(SHUTDOWN, STARTING)) return
    val startupDeadline = Deadline.fromDelay(time, config.serverMaxStartupTimeMs, TimeUnit.MILLISECONDS)
    try {
      sharedServer.startForBroker()
      // ...
      kafkaScheduler = new KafkaScheduler(config.backgroundThreads)
      // ...
      metadataCache = new KRaftMetadataCache(config.nodeId, () => raftManager.client.kraftVersion())

      // Create log manager, but don't start it because we need to delay any potential unclean shutdown log recovery
      // until we catch up on the metadata log and have up-to-date topic and broker configs.
      logManager = LogManager(config,
      // ...
      lifecycleManager = new BrokerLifecycleManager(
      // ...
      forwardingManager = new ForwardingManagerImpl(clientToControllerChannelManager, metrics)
      // ...
      // Create and start the socket server acceptor threads so that the bound port is known.
      // Delay starting processors until the end of the initialization sequence to ensure
      // that credentials have been loaded before processing authentications.
      socketServer = new SocketServer(config,
      // ...
      this._replicaManager = new ReplicaManager(
      // ...
      groupCoordinator = createGroupCoordinator()
      // ...
      transactionCoordinator = TransactionCoordinator(config, replicaManager,
      // ...
      dataPlaneRequestProcessor = new KafkaApis(
        requestChannel = socketServer.dataPlaneRequestChannel,
        forwardingManager = forwardingManager,
        replicaManager = replicaManager,
        // ...
      dataPlaneRequestHandlerPool = sharedServer.requestHandlerPoolFactory.createPool(
        config.nodeId,
        socketServer.dataPlaneRequestChannel,
        dataPlaneRequestProcessor,
        time,
        config.numIoThreads,
        "broker"
      )
      // ...
      brokerMetadataPublisher = new BrokerMetadataPublisher(config,

core/src/main/scala/kafka/server/BrokerServer.scala lines 201–531

Reading Scala — writing name = in front of an argument, as in new KafkaApis(requestChannel = …, forwardingManager = …), is a named argument. It is used to call a constructor with twenty-odd arguments without getting the order wrong. () => raftManager.client.kraftVersion() is a lambda with no arguments (Java's () -> …).

Here are the components it builds, tied to the regions of the section 1 map:

I checked the defaults in the config classes' define() calls — num.io.threads 8 (server-common/src/main/java/org/apache/kafka/server/config/ServerConfigs.java line 60, with the define on line 141), background.threads 10 (same file, lines 64 and 143), num.network.threads 3 (server/src/main/java/org/apache/kafka/network/SocketServerConfigs.java line 152, with the define on line 171).

The component map of one broker. Top to bottom is the road a request takes (B → C → D·F → E on the section 1 map); BrokerMetadataPublisher at the top right is where the ledger road comes in.

The shape SocketServer → RequestChannel → request-handler threads resembles the NIO connector in Tomcat Part 5 (Acceptor → Poller → worker thread pool). Both separate the thread that accepts connections, the threads that read and write bytes on sockets, and the threads that do the actual work. Part 4 compares where they match and where they differ.

The doors open last

The comment just above the SocketServer construction in the code above says the ports are opened early (other components need to know the bound port numbers), but request processing is delayed until the very end of initialization. Indeed, the end of the method is a series of waits:

      // Wait for this broker to contact the quorum, and for the active controller to acknowledge
      // us as caught up. It will do this by returning a heartbeat response with isCaughtUp set to
      // true. The BrokerLifecycleManager tracks this.
      FutureUtils.waitWithLogging(logger.underlying, logIdent,
        "the controller to acknowledge that we are caught up",
        lifecycleManager.initialCatchUpFuture, startupDeadline, time)
      // ...
      // We're now ready to unfence the broker. This also allows this broker to transition
      // from RECOVERY state to RUNNING state, once the controller unfences the broker.
      FutureUtils.waitWithLogging(logger.underlying, logIdent,
        "the broker to be unfenced",
        lifecycleManager.setReadyToUnfence(), startupDeadline, time)
      // ...
      val enableRequestProcessingFuture = socketServer.enableRequestProcessing(authorizerFutures)

core/src/main/scala/kafka/server/BrokerServer.scala lines 597–638

The broker starts accepting client requests only after it has waited (1) until the controller acknowledges that it has caught up with the metadata log and (2) until the controller "unfences" it. A fence is a state in which the controller marks a broker as "not trusted yet"; a fenced broker is not given leadership. Thanks to this ordering, a broker holding stale metadata never hands out wrong answers. The detailed rules for heartbeats and fencing are in Part 10.

Example 1: counting components by thread name

To see whether the component list is really alive, I started one combined-mode server and collected thread names with jstack. The number at the end of each thread name is replaced by N, and same-kind threads are counted together.

All examples in this post were run in a Docker container (Eclipse Temurin JDK 25) with Kafka 4.5.0-SNAPSHOT built as-is from the reference commit (ab1e816411). The scripts are ex/u02-threads/run.sh and ex/u02-cluster/run.sh, and the === lines in the output are headings the scripts print.

source /examples-lib/kafka.sh
kafka_single      # broker,controller in one process (PLAINTEXT 9092 / CONTROLLER 9093)
kafka-topics.sh --bootstrap-server $BOOTSTRAP --create --topic demo --partitions 3 --replication-factor 1 >/dev/null
# ...
jstack $PID > /tmp/jstack.txt
# each thread's header line in jstack starts with "name" — pull out just the names
grep '^"' /tmp/jstack.txt | sed 's/^"\([^"]*\)".*/\1/' > /tmp/names.txt
# ... drop the JVM's own threads (GC, JIT compilers, ...) and count with the trailing number replaced by N
grep -vE "$JVM" /tmp/names.txt | sed -E 's/[0-9]+$/N/' | sort | uniq -c | sort -k2

=== process.roles of node 1
process.roles=broker,controller
=== total threads: 115
=== JVM threads: 30
=== Kafka threads grouped (trailing number -> N)
      1 broker-1-lifecycle-manager-event-handler
      1 broker-1-to-controller-alter-partition-channel-manager
      1 broker-1-to-controller-forwarding-channel-manager
      1 broker-1-to-controller-heartbeat-channel-manager
      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
      1 ExpirationReaper-1-Fetch
      1 ExpirationReaper-1-Produce
      4 group-coordinator-event-processor-N
      1 group-coordinator-reaper
      1 kafka-1-metadata-loader-event-handler
      1 kafka-1-raft-io-thread
      1 kafka-1-raft-outbound-request-thread
      1 kafka-1-raft-schedulerN
      1 kafka-1-snapshot-generator-event-handler
      1 kafka-log-cleaner-thread-N
     10 kafka-scheduler-N
      1 quorum-controller-1-event-handler
      1 transaction-log-manager-N
      1 TxnMarkerSenderThread-N

I trimmed the output. The real list has 52 kinds; the 29 kinds removed above belong to features this series does not cover (share groups' share-*, ShareGroupDLQSendThread and PersisterStateManager; quotas' …-ThrottledChannelReaper-*; tiered storage's ExpirationReaper-1-RemoteFetch, and so on) or are other helper threads (metrics, controller registration, directory assignment, the DeleteRecords and AlterAcls purgatories, and so on). Of the 115 threads in total, 30 are the JVM's own (GC, JIT compilers and so on) and the remaining 85 are Kafka's. These counts can vary with the JVM and the container's CPU count (the number of GC threads follows the CPU count). kafka-1-raft-schedulerN is the real name kafka-1-raft-scheduler0 with its last digit replaced by N by the script. Grouping the names by region brings out the map:

I checked in the source where the names are made: group-coordinator-event-processor- (group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorService.java line 265), quorum-controller-<id>- (ControllerServer.scala line 242) + event-handler (server-common/src/main/java/org/apache/kafka/queue/KafkaEventQueue.java line 43), kafka-scheduler- (server-common/src/main/java/org/apache/kafka/server/util/KafkaScheduler.java line 95), and so on.

Easy to get wrong: in combined mode, request-handler thread names appear twice

The default for num.io.threads is 8, yet there are 16 data-plane-kafka-request-handler-N threads. In the full name list before trimming, data-plane-kafka-request-handler-0 through -7 each appear twice. In combined mode, BrokerServer builds a "broker" pool and ControllerServer builds a "controller" pool, each with num.io.threads threads (section 4), but the line that names the threads does not include the pool type:

    KafkaThread.daemon("data-plane-kafka-request-handler-" + id, runnables(id)).start()

core/src/main/scala/kafka/server/KafkaRequestHandler.scala line 275

So in a combined-mode server's jstack, you cannot tell from the name alone whether request-handler-3 belongs to the broker side or the controller side. You have to read down the stack to see whether it is in KafkaApis or ControllerApis. Network threads, on the other hand, carry the listener (PLAINTEXT, CONTROLLER) in their names and can be told apart. The naming is the same in 4.4.0's KafkaRequestHandler.scala.

4. What the controller holds: ControllerServer and QuorumController

The controller also has a socket server and a request-handler pool

ControllerServer.startup() is shorter than the broker's, but its skeleton is similar — it builds its own SocketServer, a dispatcher (ControllerApis) and a request-handler pool. The difference is that the thing in the middle is not ReplicaManager but QuorumController.

        new QuorumController.Builder(config.nodeId, sharedServer.clusterId).
          setTime(time).
          setThreadNamePrefix(s"quorum-controller-${config.nodeId}-").
          setConfigSchema(configSchema).
          setRaftClient(raftManager.client).
          // ...
      controller = controllerBuilder.build()
      // ...
      controllerApis = new ControllerApis(socketServer.dataPlaneRequestChannel,
        authorizerPlugin,
        quotaManagers,
        time,
        controller,
        raftManager,
        // ...
      controllerApisHandlerPool = sharedServer.requestHandlerPoolFactory.createPool(
        config.nodeId,
        socketServer.dataPlaneRequestChannel,
        controllerApis,
        time,
        config.numIoThreads,
        "controller"
      )

core/src/main/scala/kafka/server/ControllerServer.scala lines 240–305

Reading Scala — s"quorum-controller-${config.nodeId}-" is the string interpolation we saw in section 2. Chaining builder methods with a . at the end of each line works just as in Java.

The key line is the one handing raftManager.client — the Raft client built by SharedServer — to QuorumController (setRaftClient). The controller's decisions are written to the metadata log through this client.

The broker's and the controller's SocketServer are the same class. Which listeners' ports it opens depends on the kind of apiVersionManager passed in at construction:

  if (apiVersionManager.listenerType.equals(ListenerType.CONTROLLER)) {
    config.controllerListeners.foreach(createDataPlaneAcceptorAndProcessors)
  } else {
    config.dataPlaneListeners.foreach(createDataPlaneAcceptorAndProcessors)
  }

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

Reading Scala — foreach(createDataPlaneAcceptorAndProcessors) calls the method of that name (the private def on line 213) once for each listener. Passing a method name as if it were a function is the same as Java's forEach(this::createDataPlaneAcceptorAndProcessors).

That is why a combined-mode process opens two separate ports, 9092 (PLAINTEXT) and 9093 (CONTROLLER), and why Example 1 showed one set of acceptor and network threads per listener.

The controller's brain is a single thread

The doc comment of QuorumController sums up the character of the KRaft controller:

 * The node which is the leader of the metadata log becomes the active controller.  All
 * other nodes remain in standby mode.  Standby controllers cannot create new metadata log
 * entries.  They just replay the metadata log entries that the current active controller
 * has created.
 *
 * The QuorumController is single-threaded.  A single event handler thread performs most
 * operations.  This avoids the need for complex locking.
 *
 * The controller exposes an asynchronous, futures-based API to the world.  This reflects
 * the fact that the controller may have several operations in progress at any given
 * point.  The future associated with each operation will not be completed until the
 * results of the operation have been made durable to the metadata log.

metadata/src/main/java/org/apache/kafka/controller/QuorumController.java lines 160–171

Three paragraphs, three facts.

1. Even with several controllers, only one does the work. The Raft leader of the metadata log becomes the active controller; the rest are standby controllers that only replay the log.
2. Even inside the active controller, a single thread (Example 1's quorum-controller-1-event-handler) does nearly everything. Requests are put on a queue as events and that thread takes them off one at a time, so no complicated locking is needed.
3. The controller's API returns CompletableFutures, and a future completes only after the result has been made durable in the metadata log.

Many threads take requests, but only one thread makes decisions. Blue boxes are many threads, yellow boxes are where things line up, green boxes are where the result lands in the log.

Topic creation has exactly that shape:

    @Override
    public CompletableFuture<CreateTopicsResponseData> createTopics(
        ControllerRequestContext context,
        CreateTopicsRequestData request, Set<String> describable,
        boolean forwarded
    ) {
        if (request.topics().isEmpty()) {
            return CompletableFuture.completedFuture(new CreateTopicsResponseData());
        }
        return appendWriteEvent("createTopics", context.deadlineNs(),
            () -> replicationControl.createTopics(context, request, describable, forwarded));
    }

metadata/src/main/java/org/apache/kafka/controller/QuorumController.java lines 1813–1824

appendWriteEvent() puts a "write event" on the queue and returns a future. ReplicationControlManager.createTopics() inside the lambda runs on the event thread and produces a list of metadata records and a response (a ControllerResult); when the records are committed to the log, the future completes. The inside of the event queue and ControllerResult is covered in Part 10.

5. The ledger road: who decides topics, partitions and leaders, and how it spreads

Where the leader is decided

The registration that ReplicationControlManager.createTopics() builds for each partition answers "who is the first leader?":

    private static PartitionRegistration buildPartitionRegistration(
        PartitionAssignment partitionAssignment,
        List<Integer> isr
    ) {
        return new PartitionRegistration.Builder().
            setReplicas(Replicas.toArray(partitionAssignment.replicas())).
            setDirectories(Uuid.toArray(partitionAssignment.directories())).
            setIsr(Replicas.toArray(isr)).
            setLeader(isr.get(0)).
            setLeaderRecoveryState(LeaderRecoveryState.RECOVERED).
            setLeaderEpoch(0).
            setPartitionEpoch(0).
            build();
    }

metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java lines 875–888

The isr passed in here is, just above, the replica list filtered down to brokers that are currently active (partitionAssignment.replicas().stream().filter(clusterControl::isActive), lines 801–802). So the first leader is the first live broker in the replica list, and the leader epoch (a generation number that goes up by one every time the leader changes, Part 9) starts at 0. The rules for choosing which brokers get the replicas (placement) are out of scope for this post.

The decision immediately becomes records:

        records.add(new ApiMessageAndVersion(new TopicRecord().
            setName(topic.name()).
            setTopicId(topicId), (short) 0));
        // ConfigRecords go after TopicRecord but before PartitionRecord(s).
        records.addAll(configRecords);
        for (Entry<Integer, PartitionRegistration> partEntry : newParts.entrySet()) {
            int partitionIndex = partEntry.getKey();
            PartitionRegistration info = partEntry.getValue();
            records.add(info.toRecord(topicId, partitionIndex, new ImageWriterOptions.Builder(featureControl.metadataVersionOrThrow()).
                setEligibleLeaderReplicasEnabled(featureControl.isElrFeatureEnabled()).
                build()));
        }

metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java lines 860–871

One TopicRecord per topic, then (if there are configs) ConfigRecords, then one PartitionRecord per partition (info.toRecord()). The controller "creates a topic" by appending this list of records to the metadata log. It is not the controller that creates the orders-0 folder on disk, but the brokers that read these records.

The records the controller writes to create one topic. The numbers come from the actual output of example 2 in section 6.

Where the broker receives and applies the decision

On the broker side, SharedServer's MetadataLoader applies the new records to the image and calls its publisher, BrokerMetadataPublisher.onMetadataUpdate():

      // Apply topic deltas.
      Option(delta.topicsDelta()).foreach { topicsDelta =>
        try {
          // Notify the replica manager about changes to topics.
          replicaManager.applyDelta(topicsDelta, newImage)

core/src/main/scala/kafka/server/metadata/BrokerMetadataPublisher.scala lines 144–148

Reading Scala — Option(x) gives None if x is null and Some(x) otherwise. The .foreach { topicsDelta => … } that follows runs the lambda in braces only when there is a value (=> is the -> of a Java lambda).

A MetadataDelta is "the difference between the previous image and the new one". A little earlier in the same method (line 130), metadataCache.setImage(newImage) updates the cache first; here the topic part of the difference goes to ReplicaManager. ReplicaManager.applyDelta() applies only the parts that concern this broker:

  def applyDelta(delta: TopicsDelta, newImage: MetadataImage): Unit = {
    // Before taking the lock, compute the local changes
    val localChanges = delta.localChanges(config.nodeId)
    // ...
      // Handle partitions which we are now the leader or follower for.
      if (!localChanges.leaders.isEmpty || !localChanges.followers.isEmpty) {
        val lazyOffsetCheckpoints = new LazyOffsetCheckpoints(this.highWatermarkCheckpoints.asJava)
        val leaderChangedPartitions = new mutable.HashSet[Partition]
        val followerChangedPartitions = new mutable.HashSet[Partition]
        if (!localChanges.leaders.isEmpty) {
          applyLocalLeadersDelta(leaderChangedPartitions, delta, lazyOffsetCheckpoints, localChanges.leaders.asScala, localChanges.directoryIds.asScala)
        }
        if (!localChanges.followers.isEmpty) {
          applyLocalFollowersDelta(followerChangedPartitions, newImage, delta, lazyOffsetCheckpoints, localChanges.followers.asScala, localChanges.directoryIds.asScala)
        }

core/src/main/scala/kafka/server/ReplicaManager.scala lines 2369–2409

Reading Scala — new mutable.HashSet[Partition] is Scala's mutable set (scala.collection.mutable), and .asScala/.asJava wrap Java collections as Scala ones and back (scala.jdk.CollectionConverters).

delta.localChanges(config.nodeId) picks out "changes to partitions that include my node id", and among those it handles the ones where I became leader separately from those where I became follower. Partitions where it became leader get ready to accept writes; for partitions where it became follower, it attaches a fetcher thread that pulls data from the leader (Example 2's ReplicaFetcherThread-…). The inside of this is in Part 9.

The whole flow of creating one topic

But kafka-topics.sh --bootstrap-server localhost:9092 --create sends its request to a broker — so how does it reach the controller? The answer is in the match where the broker's KafkaApis.handle() dispatches by request type:

      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 ApiKeys.DELETE_TOPICS => forwardToController(request)

core/src/main/scala/kafka/server/KafkaApis.scala lines 174–191

Reading Scala — x match { case A => … case B => … } does the same job as Java's switch. Only the first matching case, from the top, runs.

The broker handles PRODUCE, FETCH and METADATA itself, while requests that change metadata, such as CREATE_TOPICS and DELETE_TOPICS, go to forwardToController(). That function calls ForwardingManager.forwardRequest() (line 144), and ForwardingManagerImpl wraps the original request in an EnvelopeRequest and sends it to the controller (server/src/main/java/org/apache/kafka/server/ForwardingManagerImpl.java line 64). The controller-side dispatcher ControllerApis.handle() has case ApiKeys.CREATE_TOPICS => handleCreateTopics(request) (core/src/main/scala/kafka/server/ControllerApis.scala line 99), which calls controller.createTopics() (line 428). The same match gathers requests only controllers receive, such as VOTE, BEGIN_QUORUM_EPOCH, BROKER_HEARTBEAT and ALTER_PARTITION (lines 96–109).

Creating one topic. Steps 1–9 are the request and response; 10–12 happen separately as each broker follows the metadata log on its own.

The observer in step 10 is a participant in a Raft quorum that follows the log without a vote. Brokers receive the metadata log as observers, not as voters in the controller quorum (Example 2's CurrentObservers). That the metadata log is also replicated with Fetch requests, like data partitions, and what the "commit" and high watermark (the boundary offset up to which replication is settled and reading is safe) in step 6 mean, are covered in Part 10 and Part 9.

Easy to get wrong: the "controller" a client sees is not the real controller

Metadata and DescribeCluster responses have a controllerId field. In the ZooKeeper days it carried the id of the real controller broker, but in this commit the broker fills it like this:

    val brokers = metadataCache.getAliveBrokerNodes(request.context.listenerName)

    trace("Sending topic metadata %s and brokers %s for correlation id %d to client %s".format(completeTopicMetadata.mkString(","),
      brokers.asScala.mkString(","), request.header.correlationId, request.header.clientId))
    val controllerId = metadataCache.getRandomAliveBrokerId.orElse(MetadataResponse.NO_CONTROLLER_ID)

core/src/main/scala/kafka/server/KafkaApis.scala lines 997–1001

It picks a random live broker and reports it as the "controller" (DescribeCluster does the same, line 2554). The source has no comment giving the reason, but it works out: a KRaft controller may be a separate node that does not open a client listener, and any broker you send a metadata-changing request to will forward it to the controller anyway. In fact the admin client (Admin) sends CreateTopics straight to this "controller" broker (new Call("createTopics", deadline, new ControllerNodeProvider()), clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java line 1924), so the "broker 1" slot in the topic-creation diagram above may be filled by a randomly chosen broker rather than the bootstrap broker. In Example 2 we see Admin.describeCluster().controller() return a broker id rather than controller node 100. To see the real controller quorum, ask the controllers directly with kafka-metadata-quorum.sh.

6. The data road: how a client finds the leader

The Metadata request — ask any broker

Producers and consumers know only the few brokers listed in bootstrap.servers. They learn the rest through Metadata requests. When it is time to refresh metadata, DefaultMetadataUpdater.maybeUpdate() inside the client's NetworkClient picks the broker to ask like this:

            // Beware that the behavior of this method and the computation of timeouts for poll() are
            // highly dependent on the behavior of leastLoadedNode.
            LeastLoadedNode leastLoadedNode = leastLoadedNode(now);
            // ...
            return maybeUpdate(now, leastLoadedNode.node());

clients/src/main/java/org/apache/kafka/clients/NetworkClient.java lines 1473–1493

leastLoadedNode() is, as the name says, "the broker with the fewest requests currently outstanding". Every broker holds the same metadata (in its own KRaftMetadataCache), so you can ask any of them. The response contains the list of live brokers and, per partition, the leader, replicas and ISR (handleTopicMetadataRequest(), lines 891–1013).

If KafkaProducer.send() has no metadata for the topic yet, it asks for a refresh with metadata.requestUpdateForTopic(topic) inside waitOnMetadata(), wakes the Sender thread, and waits (clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java lines 1327–1330). The max.block.ms setting you know from your app is the upper bound for this wait (Part 5).

The Produce request — always to that partition's leader

RecordAccumulator.ready(), where the Sender thread picks batches to send, looks up the leader in the metadata for each batch:

                // We didn't get leader when queue sizes is null so get it now
                if (queueSizes == null) {
                    leader = cluster.leaderFor(part);
                }
                // Check batch ready as a short-circuit to avoid expensive per-batch computation.
                checkBatchReady = leader != null && !readyNodes.contains(leader) && !isMuted(part);

clients/src/main/java/org/apache/kafka/clients/producer/internals/RecordAccumulator.java lines 820–825

If the leader is unknown (leader == null), the batch is not ready to send. Ready batches are grouped by leader broker and become one Produce request to that broker (the detailed rules of ready() and drain(), and linger.ms, are in Part 5).

What if the leader changed in the meantime? A broker that is not the leader answers with a NOT_LEADER_OR_FOLLOWER error, and when Sender gets this error it asks for fresh metadata:

            if (error.exception() instanceof InvalidMetadataException) {
                // ...
                metadata.requestUpdate(false);
            }

clients/src/main/java/org/apache/kafka/clients/producer/internals/Sender.java lines 713–731

NotLeaderOrFollowerException is a subclass of InvalidMetadataException (public class NotLeaderOrFollowerException extends InvalidMetadataException, NotLeaderOrFollowerException.java line 31). In the trimmed part (lines 722–729), if the response carried new leader information, that is recorded as well. How retries relate to delivery.timeout.ms is in Part 5.

Metadata from any broker, Produce to the partition's leader. The leader layout (steps 4–6) is from one run of Example 2 and differs from run to run.

Example 2: checking the map on a 1-controller + 3-broker cluster

This time I split the roles: one controller (node 100, process.roles=controller) and three brokers (nodes 1, 2, 3, process.roles=broker) in one container, then created a topic orders with 3 partitions and RF 3, and checked:

• The quorum state: who is a voter and who is an observer
• Leader/Replicas/Isr from --describe, and the folders on disk
• The map the producer got from Metadata, and which broker each Produce request actually reached (by turning the request logger kafka.request.logger up to DEBUG at runtime and reading kafka-request.log)
• Which threads the controller process and a broker process each have

The Java part prints the Metadata response through the producer's partitionsFor(), sends one record to each partition, then prints the "controller" the broker reports through Admin.describeCluster().

// An example written for this post (not part of Kafka's source)
// Prints the partition map (leader, replicas, ISR) the producer got from a Metadata response,
// then sends one record to each partition to see which broker (leader) it actually went to.
// ... (imports omitted)
public class Leaders {
    public static void main(String[] args) throws Exception {
        Properties p = new Properties();
        // only broker 1 as the bootstrap address — the other brokers are learned from the Metadata response
        p.put("bootstrap.servers", "localhost:9092");
        p.put("key.serializer", StringSerializer.class.getName());
        p.put("value.serializer", StringSerializer.class.getName());

        try (KafkaProducer<String, String> producer = new KafkaProducer<>(p)) {
            // partitionsFor() sends a Metadata request and waits for the response
            System.out.println("--- partitionsFor(\"orders\") : from the Metadata response");
            for (PartitionInfo info : producer.partitionsFor("orders")) {
                System.out.printf("partition %d  leader=%s  replicas=%s  isr=%s%n",
                        info.partition(), info.leader().idString(),
                        ids(info.replicas()), ids(info.inSyncReplicas()));
            }
            System.out.println("--- send one record to each partition");
            for (int part = 0; part < 3; part++) {
                RecordMetadata md = producer.send(new ProducerRecord<>("orders", part, "k" + part, "v" + part)).get();
                System.out.printf("partition %d  offset=%d%n", md.partition(), md.offset());
            }
        }

        try (Admin admin = Admin.create(p)) {
            // in KRaft, a broker reports a randomly chosen live broker instead of the real controller
            System.out.println("--- describeCluster() via broker 1");
            System.out.println("nodes      = " + admin.describeCluster().nodes().get().stream().map(Node::idString).sorted().toList());
            System.out.println("controller = " + admin.describeCluster().controller().get().idString());
        }
    }
    // ... ids(): turns a Node[] into a string like "[1,2,3]"
}

Here is the output. The cluster id, directory ids, topic id, high watermark value, leader layout and the randomly chosen "controller" differ from run to run.

=== process.roles
node 100: process.roles=controller
node 1: process.roles=broker
node 2: process.roles=broker
node 3: process.roles=broker
=== quorum (kafka-metadata-quorum.sh describe --status)
ClusterId:              77vznZ5CSbmKGGPryO4TLA
LeaderId:               100
LeaderEpoch:            1
HighWatermark:          60
CurrentVoters:          [{"id": 100, "directoryId": "Kauh39PwTXmTmoIsqH798Q", "endpoints": ["CONTROLLER://localhost:9093"]}]
CurrentObservers:       [{"id": 2, "directoryId": "Ezc48wiVShBbWTUeK_ivag"}, {"id": 3, "directoryId": "0rLPoqAnudMkqyjVp2qgAA"}, {"id": 1, "directoryId": "P6YemdSO2aWPDbGmEWH-wA"}]
=== create topic orders (3 partitions, RF=3)
Topic: orders	TopicId: scuPXA0ORiq9ntTcjVSiUw	PartitionCount: 3	ReplicationFactor: 3	Configs: min.insync.replicas=1
	Topic: orders	Partition: 0	Leader: 3	Replicas: 3,1,2	Isr: 3,1,2	Elr: 	LastKnownElr: 
	Topic: orders	Partition: 1	Leader: 1	Replicas: 1,2,3	Isr: 1,2,3	Elr: 	LastKnownElr: 
	Topic: orders	Partition: 2	Leader: 2	Replicas: 2,3,1	Isr: 2,3,1	Elr: 	LastKnownElr: 
=== partition folders on each node's disk
node 1: __cluster_metadata-0 orders-0 orders-1 orders-2 
node 2: __cluster_metadata-0 orders-0 orders-1 orders-2 
node 3: __cluster_metadata-0 orders-0 orders-1 orders-2 
node 100: __cluster_metadata-0 
=== Java client
--- partitionsFor("orders") : from the Metadata response
partition 0  leader=3  replicas=[3,1,2]  isr=[3,1,2]
partition 1  leader=1  replicas=[1,2,3]  isr=[1,2,3]
partition 2  leader=2  replicas=[2,3,1]  isr=[2,3,1]
--- send one record to each partition
partition 0  offset=0
partition 1  offset=0
partition 2  offset=0
--- describeCluster() via broker 1
nodes      = [1, 2, 3]
controller = 2
=== which broker received the PRODUCE requests (kafka-request.log)
node 1 received PRODUCE for orders-1
node 2 received PRODUCE for orders-2
node 3 received PRODUCE for orders-0
--- one PRODUCE line (request part only)
"request":{"transactionalId":null,"acks":-1,"timeoutMs":30000,"topicData":[{"topicId":"scuPXA0ORiq9ntTcjVSiUw","partitionData":[{"index":1,"recordsSizeInBytes":72}]}]},"response"
=== who has which threads (count per process)
thread name pattern                       node100    node1
quorum-controller-                              1        0
-raft-io-thread                                 1        1
metadata-loader-event-handler                   1        1
data-plane-kafka-request-handler-               8        8
data-plane-kafka-network-thread-                3        3
group-coordinator-event-processor-              0        4
TxnMarkerSenderThread                           0        1
ReplicaFetcherThread-                           0        2
kafka-log-cleaner-thread-                       0        1
broker-.*-lifecycle-manager                     0        1
stopped

Reading the output alongside the sections of this post:

1. Quorum: the only voter (CurrentVoters) is controller 100, and it is the leader (LeaderId: 100) — that is, the active controller (section 4). Brokers 1, 2 and 3 follow the metadata log as observers (CurrentObservers) (section 5).
2. Leader = first in the replica list: for all three partitions, Leader equals the first value of Replicas. That is section 5's setLeader(isr.get(0)). Because the replica list order shifts by one for each partition, the leaders ended up spread evenly across the three brokers.
3. Disk: the data folders orders-N exist only on the three brokers; controller 100 has only __cluster_metadata-0. The brokers also have a metadata log folder — the copy they receive as observers.
4. Produce goes to the leader: the producer was given only broker 1 as its bootstrap address, yet the Produce for orders-0 went to broker 3, its leader, and orders-2 to broker 2 (section 6). You can also see that the request body identifies the partition by topicId rather than by topic name, and that acks is -1 (= all, the producer default in this commit — ProducerConfig.java lines 430–433). The byte shape of requests is in Part 3, and the other fields of the request log (queue time, processing time) in Part 4.
5. The "controller" is a broker id: describeCluster().controller() returned 2, not 100 (section 5's "Easy to get wrong"). Other runs returned 1 and 3.
6. Who has what: the quorum-controller- thread exists only on the controller; the group coordinator, transaction markers, replica fetchers, log cleaner and lifecycle manager exist only on the broker. raft-io-thread and metadata-loader are on both — they are components SharedServer (section 2) builds regardless of role. Broker 1 has 2 ReplicaFetcherThread-s because the leaders it follows as a follower are on two brokers, 2 and 3 (thread names have the form ReplicaFetcherThread-<fetcherId>-<leader broker id>, ReplicaFetcherManager.scala line 42, with num.replica.fetchers defaulting to 1). The controller, too, has 8 request-handler threads and 3 network threads (section 4).

Example 2's layout. Only the replication arrows for orders-0 are drawn (orders-1 and orders-2 look the same, each pointing at its own leader).

Easy to get wrong: the leader layout changes from run to run

The first time I ran Example 2, the leader of orders-0 was broker 1; the second time, broker 2; the third time (the output above), broker 3. Which broker the replica list starts from differed from run to run. Still, in all three runs "leader = first in the replica list" and "Produce goes to that leader" held. The placement rules themselves are out of scope for this series.

7. Which way the modules depend: brokers use clients too

The direction build.gradle gives

We looked at the folder map in Part 1; now let's see which of those folders (Gradle modules) depend on which. core's dependency list shows the top layer:

  dependencies {
    releaseOnly log4jReleaseLibs
    // `core` is often used in users' tests, define the following dependencies as `api` for backwards compatibility
    // even though the `core` module doesn't expose any public API
    api project(':clients')
    api libs.scalaLibrary
    // ...
    implementation project(':server-common')
    implementation project(':group-coordinator:group-coordinator-api')
    implementation project(':group-coordinator')
    implementation project(':transaction-coordinator')
    implementation project(':metadata')
    implementation project(':storage:storage-api')
    // ...
    implementation project(':tools:tools-api')
    implementation project(':raft')
    implementation project(':storage')
    implementation project(':server')
    implementation project(':coordinator-common')
    implementation project(':share-coordinator')
    implementation project(':mirror-coordinator')

build.gradle lines 1175–1198

core depends on nearly every server module. That is natural: BrokerServer and ControllerServer, which assemble brokers and controllers, live here. Reading the other modules' dependencies the same way and drawing them as arrows gives this (test dependencies and *-api modules left out):

Module dependency direction (an arrow means "depends on"). Every arrow in the figure is in build.gradle, but to keep the figure simple some skipping arrows (for example core → storage, metadata → clients) are left out.

Two things stand out. First, clients is at the very bottom and depends on no server module. That is why an app can take just kafka-clients.jar. Second, as the raft → storage arrow shows, the metadata log sits on the same storage layer as data logs. The Raft log implementation (KafkaRaftLog) uses storage's UnifiedLog (Part 10).

The same Selector, the same record classes

"Brokers use clients too" is not just talk. Look at the selector a broker network thread (Processor) creates to handle sockets:

  protected[network] def createSelector(channelBuilder: ChannelBuilder): KSelector = {
    channelBuilder match {
      case reconfigurable: Reconfigurable => config.addReconfigurable(reconfigurable)
      case _ =>
    }
    new KSelector(
      maxRequestSize,
      // ...
      logContext)
  }

core/src/main/scala/kafka/network/SocketServer.scala lines 846–864

Reading Scala — KSelector is imported under a new name on line 36 of this file, import org.apache.kafka.common.network.{…, Selector => KSelector} (so it is not confused with Java NIO's java.nio.channels.Selector). case _ => means "in every other case, do nothing". protected[network] restricts visibility to the network package.

org.apache.kafka.common.network.Selector — a class from the clients module. The producer uses the same class (selector = new Selector(config.getLong(CommonClientConfigs.CONNECTIONS_MAX_IDLE_MS_CONFIG), …, clients/src/main/java/org/apache/kafka/clients/ClientUtils.java line 283). Records are the same story: the broker's UnifiedLog imports clients' MemoryRecords, FileRecords and RecordBatch (storage/…/UnifiedLog.java lines 36–43). The design in which the broker reads the producer's record-batch format with the same classes and writes it to disk almost unchanged is covered in Part 6.

Counting, in the reference commit, the src/main files that import org.apache.kafka.common.… (that is, clients' shared packages) shows this direction in numbers too:

cd /d/openSource/kafka-ab1e816411
for m in core server storage raft metadata group-coordinator; do
  files=$(git ls-files "$m/src/main" | grep -E '\.(java|scala)$')
  echo "$m total=$(echo "$files" | wc -l)" \
       "common=$(echo "$files" | xargs grep -l '^import org\.apache\.kafka\.common\.' | wc -l)" \
       "record=$(echo "$files" | xargs grep -l '^import org\.apache\.kafka\.common\.record\.' | wc -l)" \
       "network=$(echo "$files" | xargs grep -l '^import org\.apache\.kafka\.common\.network\.' | wc -l)"
done

core total=67 common=61 record=11 network=15
server total=172 common=122 record=6 network=13
storage total=123 common=78 record=20 network=0
raft total=91 common=61 record=18 network=14
metadata total=179 common=104 record=4 network=2
group-coordinator total=112 common=69 record=3 network=0

Going the other way, the number of files in clients/src/main that import the kafka., org.apache.kafka.storage., raft., metadata., coordinator. or network. packages was 0. The heaviest users of the record classes are storage (logs) and raft (the metadata log), and the network classes are used by core, server and raft, which handle sockets directly.

Easy to get wrong: package names don't tell you the module

By its name, org.apache.kafka.server.authorizer.Authorizer looks like it belongs to the server module, but it lives in clients/src/main/java/org/apache/kafka/server/authorizer/. clients contains server/authorizer, server/policy, server/quota and server/telemetry: plugin interfaces that users implement and plug into the broker, so they sit in the public-API clients jar. Conversely, core's Scala code is in the kafka.server package, and the server module's Java code is in org.apache.kafka.server. When looking for a class, don't guess from the package name — git ls-files | grep '/Name\.\(java\|scala\)$' is faster.

What this post left out

Things left out to keep the map to one page:

• How KafkaRaftServer.initializeLogDirs() checks meta.properties, and kafka-storage.sh format. core/src/main/scala/kafka/server/KafkaRaftServer.scala lines 131–188
• Replica placement rules (which brokers get the replicas). metadata/src/main/java/org/apache/kafka/metadata/placement/
• Broker registration, heartbeat and fencing rules. server/src/main/java/org/apache/kafka/server/BrokerLifecycleManager.java — Part 10
• Controller registration (KIP-919, ControllerRegistrationManager) and adding/removing controller quorum voters (ADD_RAFT_VOTER)
• Dynamic configuration (DynamicBrokerConfig, DynamicConfigPublisher)
• Left out of the whole series: Kafka Streams (streams/), Kafka Connect (connect/), MirrorMaker, share groups (share-coordinator/, the share-* threads in Example 1), tiered storage (RemoteLogManager), security (SASL, TLS, ACLs, Authorizer), quotas (Example 1's ThrottledChannelReaper-*), the inside of the admin client Admin (the examples only use it as a tool), JBOD and ReplicaAlterLogDirsThread

Summary

What's next

We fold the map and start with the language every arrow on it shares. The next post is the arrow between A and B on the section 1 map — the wire protocol clients and brokers speak. We'll see where names like "requestApiKeyName":"PRODUCE", "topicData" and "partitionData" in Example 2's request log come from, how request and response classes are generated from JSON schemas at build time, and why a client agrees on versions with ApiVersions right after connecting — with an example that sends serialized bytes straight down a socket.

Digging into Kafka's Source 3: The Wire Protocol — Requests and Responses Generated from JSON Schemas

The earlier series spread a big-picture map at the same point. In Tomcat Part 2, a single server.xml became a tree of Server, Service, Engine and Host objects, and Coyote (which accepts connections) and Catalina (which calls servlets) were joined by a single Adapter. In Kafka, one config line (process.roles) picks two parts — the broker and the controller — and the two are joined by SharedServer's metadata log. Put side by side with the core/MPM/module map of httpd Part 3 and the module map of Spring Part 2, Kafka is the only one of the four that puts the shape of a distributed system — "several servers sharing the keeping of one ledger" — at the center of its map. The nginx series, MariaDB series and Gradle series, being written at the same time, also each start by spreading out a map in their Part 1.