apache/kafka

core (broker)

Active contributors: Ken Huang, PoAn Yang, Apoorv Mittal, TengYao Chi, David Jacot

Purpose

The core/ module is the Scala broker. It owns the request pipeline (SocketServer → KafkaApis), partition replication (ReplicaManager and the fetcher threads), log management (LogManager plus the Java-side log code in storage/), and the in-process orchestration that boots a node as a broker, controller, or both. Almost every RPC ends up dispatched through this module.

Directory layout

core/src/main/scala/kafka/
├── Kafka.scala                      ← main(); parses args, builds KafkaRaftServer, runs it
├── server/
│   ├── KafkaRaftServer.scala        ← combined-mode entry; wraps a SharedServer
│   ├── SharedServer.scala           ← infra shared between broker and controller roles
│   ├── BrokerServer.scala           ← broker role: starts ReplicaManager, log layer, coordinators
│   ├── ControllerServer.scala       ← controller role: starts KafkaRaftClient, QuorumController
│   ├── KafkaApis.scala              ← (~5K lines) request dispatch — every API key handled here
│   ├── KafkaConfig.scala            ← server-side config types and validation
│   ├── ReplicaManager.scala         ← (~3.5K lines) read/write path, ISR tracking
│   ├── ReplicaFetcherThread.scala / Manager / AbstractFetcherThread.scala
│   ├── DelayedFetch.scala / DelayedProduce.scala / DelayedJoin.scala / ...
│   ├── ConfigAdminManager.scala     ← AlterConfigs / DescribeConfigs handling
│   ├── ForwardingManager.scala      ← brokers forward writes to the controller
│   └── metadata/                    ← BrokerMetadataPublisher, listener for metadata events
├── network/                         ← SocketServer, Acceptor, Processor (NIO accept/read loop)
├── log/                             ← LogManager.scala + Scala adapters around storage/ Java code
├── cluster/                         ← Partition.scala, Broker.scala, EndPoint.scala
├── coordinator/                     ← Scala glue around the Java group/transaction coordinators
├── raft/                            ← Scala adapters around the Java raft module
├── tools/                           ← legacy CLI bits still in Scala (most CLIs have moved to tools/)
├── metrics/                         ← KafkaMetricsGroup, JMX bridges
└── utils/                           ← Logging trait, ZkUtils replacement helpers

core/src/test/scala/ mirrors the layout for unit tests; integration tests under core/src/test/scala/integration/ use the in-process KafkaClusterTestKit.

Key abstractions

Type	File	Purpose
`Kafka` object	`core/src/main/scala/kafka/Kafka.scala`	Main entry point invoked by `kafka-server-start.sh`.
`KafkaRaftServer`	`core/src/main/scala/kafka/server/KafkaRaftServer.scala`	Top-level lifecycle for combined / broker / controller roles.
`SharedServer`	`core/src/main/scala/kafka/server/SharedServer.scala`	Infrastructure shared across roles: log dirs, metadata loader, raft client, etc.
`BrokerServer`	`core/src/main/scala/kafka/server/BrokerServer.scala`	Broker role: starts request handlers, coordinators, replica manager.
`ControllerServer`	`core/src/main/scala/kafka/server/ControllerServer.scala`	Controller role: starts `KafkaRaftClient`, `QuorumController`.
`SocketServer` / `Acceptor` / `Processor`	`core/src/main/scala/kafka/network/SocketServer.scala`	NIO accept loop and per-connection processor threads.
`KafkaRequestHandler` / `RequestChannel`	`core/src/main/scala/kafka/server/KafkaRequestHandler.scala`	Pulls requests from the channel and runs them on the request handler thread pool.
`KafkaApis`	`core/src/main/scala/kafka/server/KafkaApis.scala`	One method per API key (`handleProduceRequest`, `handleFetchRequest`, …).
`ReplicaManager`	`core/src/main/scala/kafka/server/ReplicaManager.scala`	Append, fetch, ISR maintenance, partition state.
`Partition`	`core/src/main/scala/kafka/cluster/Partition.scala`	Per-partition state on a single broker (leader/follower, ISR set).
`LogManager`	`core/src/main/scala/kafka/log/LogManager.scala`	Owns all `UnifiedLog` instances; rolling, retention, cleaner.
`KafkaScheduler`	`core/src/main/scala/kafka/utils/KafkaScheduler.scala`	Background-task scheduler used by the broker.

Request pipeline

graph LR
    NIC[Client TCP] --> A[Acceptor<br/>kafka.network]
    A -->|round-robin| P1[Processor 1] & P2[Processor 2] & PN[Processor N]
    P1 & P2 & PN -->|RequestChannel.send| RC[(RequestChannel<br/>queue)]
    RC --> KRH[KafkaRequestHandler pool]
    KRH --> KA[KafkaApis.handle]
    KA -->|Produce/Fetch| RM[ReplicaManager]
    KA -->|JoinGroup, Heartbeat, OffsetCommit| GC[Group coordinator]
    KA -->|InitProducerId, AddOffsetsToTxn| TC[Transaction coordinator]
    KA -->|CreateTopics, AlterConfigs| FW[ForwardingManager → controller]
    RM --> Log[(UnifiedLog)]
    KA -->|response| RC
    RC --> P1

KafkaApis is structurally a giant match on the API key; each branch validates the request, performs authorization (AuthHelper), and delegates to a subsystem. For writes, it calls into ReplicaManager.appendRecords; for reads, ReplicaManager.fetchMessages; for group operations, the relevant coordinator. Slow operations (delayed fetch waiting for min.bytes, delayed produce waiting for acks=all) park on per-purpose DelayedOperationPurgatory instances.

Replication

Followers keep up with leaders by running ReplicaFetcherThread instances, one per leader broker, all managed by ReplicaFetcherManager. The fetcher loop:

sequenceDiagram
    participant F as ReplicaFetcherThread (follower)
    participant RM as ReplicaManager (leader)
    participant L as UnifiedLog (leader)
    F->>RM: FetchRequest(version, fetchOffset[partitions])
    RM->>L: read from local log (respecting HW)
    L-->>RM: Records
    RM-->>F: FetchResponse with records + HW
    F->>F: append to local log
    F->>F: advance LEO; periodically advance HW

ISR maintenance: the leader watches each follower's reported LEO and, if it drops behind by more than replica.lag.time.max.ms, removes it from the ISR via an AlterPartition RPC sent to the controller. The controller authoritatively updates the metadata log, brokers see the update via MetadataLoader, and the local Partition state is reconciled.

ReplicaAlterLogDirsThread is a sibling that moves a replica between log directories on the same broker, used by JBOD reassignment.

Boot sequence (broker role)

sequenceDiagram
    participant Main
    participant SS as SharedServer
    participant BS as BrokerServer
    participant ML as MetadataLoader
    participant RM as ReplicaManager

    Main->>SS: startForBroker()
    SS->>SS: open log dirs, start KafkaRaftClient (observer)
    SS->>ML: register publishers
    Main->>BS: startup()
    BS->>RM: new ReplicaManager
    BS->>BS: build group/share/transaction coordinators
    BS->>BS: register BrokerMetadataPublisher
    ML-->>BS: replay metadata records
    BS-->>BS: PartitionsCreated → ReplicaManager.applyDelta
    BS->>BS: bind SocketServer ports, advertise to controller

BrokerMetadataPublisher (core/src/main/scala/kafka/server/metadata/BrokerMetadataPublisher.scala) listens for MetadataDelta events and translates them into local actions (create partitions, update topic config, install ACLs, refresh quotas).

Configuration

KafkaConfig is the top-level broker config; it is built from the union of declared ConfigDef entries spread across ReplicationConfigs, LogConfig, KRaftConfigs, ServerConfigs, and so on. Dynamic config changes (per-broker or cluster-wide) are layered on top by DynamicBrokerConfig (core/src/main/scala/kafka/server/DynamicBrokerConfig.scala), which subscribes to BrokerConfig records in the metadata log and reconfigures live components via the Reconfigurable SPI.

Authorization and ACLs

AuthHelper (core/src/main/scala/kafka/server/AuthHelper.scala) centralizes authorization checks. The pluggable Authorizer SPI lives in clients/src/main/java/org/apache/kafka/server/authorizer/; the default StandardAuthorizer (metadata/src/main/java/org/apache/kafka/metadata/authorizer/StandardAuthorizer.java) sources ACLs from the metadata image. See features/acls-and-security.md.

Metrics

The broker exposes a few thousand JMX metrics. Local metric producers are subclasses of KafkaMetricsGroup (core/src/main/scala/kafka/metrics/KafkaMetricsGroup.scala), and the standard names are documented in docs/ops.html. The same org.apache.kafka.common.metrics.Metrics registry that backs the clients also backs the broker.

Entry points for modification

Add an RPC handler: edit KafkaApis.scala and add a handle<NewApi> method. The request/response classes are generated from a schema in clients/src/main/resources/common/message/. KIP required.
Change replication semantics: start in Partition.scala and ReplicaManager.scala. Anything that affects the leader epoch/HW must be reflected in the inter-broker FetchRequest/AlterPartition protocol and tested with ReplicaFetcherThreadTest + integration tests.
Add a broker config: declare it in the appropriate *Configs Java class under server-common//clients/, wire it into KafkaConfig, and document it. KIP required for any user-visible config.
Change boot order or component initialization: BrokerServer.scala and SharedServer.scala. Be careful — these orderings are subtle and tested by integration tests in kafka.testkit.

Architecture — top-level diagrams.
Modules: clients — wire protocol used at the broker boundary.
Modules: storage — on-disk log code that LogManager calls.
Modules: raft and Modules: metadata — control plane.
Features: KRaft mode — broker / controller / combined roles.
Features: Replication.

Built by Factory AutoWiki from public repository content. It is a generated preview for codebase exploration, not source-maintained documentation.