Apache Kafka — Architecture, Message Flow, and CLI Reference

1. Architecture

A Kafka cluster is a set of brokers that store messages in append-only logs. Each log is a partition, and partitions are grouped into named topics. Every partition has one leader broker and zero or more follower replicas; producers always send to the leader for a given partition, and consumers normally fetch from the leader (follower fetching via KIP-392 / rack-aware reads can serve fetches from in-sync replicas when enabled). The set of replicas that are caught up with the leader is the in-sync replica set (ISR). Cluster metadata (which broker leads which partition, which replicas are in-sync, topic configs, ACLs) is managed by the controller — in modern Kafka, this is a KRaft quorum running on co-located or dedicated controller nodes; in older deployments, it is an external ZooKeeper ensemble.

Solid arrows: producer → partition leader and leader → consumer fetch. Dashed arrows: leader → follower replication (only ISR members are eligible for leader election on failure).

Core concepts in one line each

• Broker — a Kafka server process; holds partition replicas on local disk and serves produce / fetch requests.
• Topic — a named, partitioned, append-only log. Retention is time-based or size-based; compacted topics keep the latest value per key.
• Partition — the unit of parallelism, ordering, and replication. Order is guaranteed within a partition, never across partitions.
• Leader / Follower — one replica per partition is the leader; followers replicate from it. Reads and writes go to the leader by default.
• ISR (in-sync replicas) — replicas that have caught up to the leader. acks=all + min.insync.replicas defines the durability contract.
• Producer — picks a partition (by key hash, sticky partitioner, explicit assignment, or a custom partitioner), batches records, and waits for the configured acks before considering a send successful.
• Consumer group — a set of consumers that share partition assignments via the group coordinator. Each partition is consumed by exactly one member of a group at a time.
• Offset — the monotonically increasing position within a partition. Consumers commit progress to the __consumer_offsets internal topic.
• Controller — broker(s) that own cluster metadata. In KRaft mode this is a Raft quorum of dedicated or co-located controllers; in legacy mode it lives in ZooKeeper.

2. Message Flow

The end-to-end path of a single record with acks=all, two-replica replication, and a consumer group that commits offsets after processing.

Blue arrows: requests. Green arrows: successful responses / acks. HW (high-water mark) is the minimum log-end offset across the in-sync replicas — i.e., HW = min(LEO of ISR) — and consumers can only read up to it.

What the steps guarantee

• Step 1–6 (produce path with acks=all) — the producer's send returns success only after every in-sync replica has appended the record. With min.insync.replicas=2, a single broker loss does not lose acknowledged data.
• Step 5 (high-water mark) — readers only see offsets at or below the HW. Records above HW are written to disk but not yet visible to consumers, which prevents a consumer from reading data that might be lost on leader failover.
• Step 7–8 (consumer poll) — order is preserved within a partition. Two consumers in the same group never receive the same partition, so each record is processed by exactly one consumer per group.
• Step 9–10 (offset commit) — committing after processing gives at-least-once delivery (a crash between processing and commit re-processes records). Committing before processing gives at-most-once. Exactly-once across produce + consume requires the transactional API (enable.idempotence=true, transactional.id, isolation.level=read_committed).

3. Exactly-Once Semantics

Kafka's exactly-once story is built out of two independent mechanisms layered on top of each other: an idempotent producer that eliminates duplicates from producer retries within a single partition, and a transactional API that ties multiple partition writes and a consumer-offset commit into a single atomic unit. The transactional API enables the canonical consume → process → produce pattern (the basis of Kafka Streams' EOS guarantee) where the input offset commit and the output records become visible to downstream consumers together or not at all. Note that this guarantee applies inside Kafka — Kafka transactions do not provide exactly-once guarantees for external systems such as relational databases.

Idempotent producer

When you set enable.idempotence=true, the producer is assigned a 64-bit Producer ID (PID) by the broker at first contact and a per-partition monotonically increasing sequence number is attached to every record batch. The leader broker remembers the last sequence number it accepted from a (PID, partition) pair; any batch arriving with a sequence number that has already been committed (because the producer retried after a transient ack failure) is silently de-duplicated, and any out-of-order sequence is rejected with OutOfOrderSequenceException. The guarantee is scoped to a single producer session and a single partition.

Transactional API

Setting transactional.id (in addition to enable.idempotence=true) promotes the producer to a transactional one and pins it to a transaction coordinator — the broker that leads the __transaction_state partition corresponding to the transactional.id hash. The coordinator hands out a producer epoch that fences any older instance of the same transactional.id (so a zombie producer that comes back after a long pause cannot publish into a transaction owned by its successor). The transaction lifecycle is a two-phase commit:

• BeginTxn — producer signals "I'm starting a transaction". No broker-side write yet.
• AddPartitionsToTxn — for every new partition the producer touches, it tells the coordinator. The coordinator records this in __transaction_state so it knows which partitions need a commit/abort marker at the end.
• Send records — the producer writes data records as usual, but each batch carries the (PID, epoch, txn-flag). The records are appended to the data partitions before commit; consumers using read_committed will not see them yet.
• SendOffsetsToTxn (consume-process-produce) — the producer tells the coordinator "commit these consumer offsets atomically with the transaction". The coordinator forwards the offset commit to the group coordinator inside the transaction.
• CommitTxn / AbortTxn — the coordinator writes a PREPARE_COMMIT (or PREPARE_ABORT) record to __transaction_state, then dispatches control records (commit or abort markers) to every partition the transaction touched. Only after all markers are appended does the coordinator write COMPLETE_COMMIT. Consumers using isolation.level=read_committed follow the markers to filter out aborted batches and only return committed ones, up to the last stable offset (LSO) — the offset before the earliest still-open transaction.

Blue: client request flow. Orange: coordinator-internal control writes that drive the commit decision. Green: commit markers + the LSO advance that makes data visible to read_committed consumers. Until step 10 lands on every touched partition, the data records from steps 5–6 sit in the log but are invisible to read_committed.

Key knobs and properties

• Producer — enable.idempotence=true, transactional.id=<stable-id>, acks=all (forced by idempotence), max.in.flight.requests.per.connection ≤ 5 (forced).
• Consumer — isolation.level=read_committed. With read_uncommitted (the default) you see aborted batches too.
• Broker — transaction.state.log.replication.factor, transaction.state.log.min.isr, transaction.max.timeout.ms. The __transaction_state topic is created automatically.
• Failure mode — if the producer crashes mid-transaction, the coordinator times out the transaction and writes an AbortTxn marker, so consumers never see the half-written batches. A zombie producer with a stale epoch is rejected on its next request.

4. Memory / CPU / I/O Pipeline

A Kafka broker is essentially a network-attached append-only log that aggressively uses the kernel page cache instead of a JVM-resident cache. The throughput numbers Kafka is known for come from three architectural choices: sequential disk I/O on the produce path, page-cache reads served via sendfile() zero-copy on the consume path, and a thread pool design that decouples slow network I/O from CPU-bound request processing.

Memory layout of a broker process

• JVM heap (commonly configured between several GB and tens of GB depending on workload) — controller / metadata cache, request queues, in-flight produce buffers, request-handler local state, index entries cached in heap. Intentionally kept small relative to RAM so the kernel can use the rest as page cache.
• Off-heap (direct buffers) — Java NIO buffers used by the network layer and by FileChannel reads/writes. Sized by -XX:MaxDirectMemorySize.
• Page cache (all remaining RAM) — OS-owned, holds recently written log segments. Both producers and consumers hit this rather than going to disk. A reader at the tail of the log (lag ≈ 0) almost never touches a spinning platter; it reads the same page the producer just wrote.
• Disk — log segments stored as flat append-only files under log.dirs. The OS flushes dirty pages on its own schedule (controlled by vm.dirty_* sysctls). By default Kafka does not fsync every record (broker-driven flushing can be opted into via log.flush.interval.*, but the path is rarely used in practice); durability comes from ISR replication, not from waiting for disk.

CPU / threading model

• Network threads (num.network.threads, default 3) — non-blocking epoll-style loop. One acceptor binds the listener, the others read framed requests off sockets and enqueue them onto a request channel. No per-connection thread.
• Request handler / IO threads (num.io.threads, default 8) — pull from the request channel and execute the actual work against ReplicaManager / LogManager / GroupCoordinator / TransactionCoordinator. This is where the CPU spends most of its time on the produce path (CRC, compression, log append) and on the consume path (offset lookup, fetch session bookkeeping).
• Replica fetcher threads — follower brokers run dedicated fetcher threads that issue FetchRequest against partition leaders to replicate. ISR membership is driven by how fresh each follower's fetch position is relative to the leader's log-end offset.
• Background threads — log cleaner (compaction), log retention deleter, group coordinator heartbeat, transaction coordinator expiration, controller event thread (in KRaft mode the controllers run a separate Raft event loop).

Network traffic

• Client ↔ broker — long-lived TCP connections, one per broker the client talks to. Producer sends ProduceRequest per batch; consumer sends FetchRequest per partition-set per poll; both multiplex over the same connection. OffsetCommitRequest and group coordination (JoinGroup, SyncGroup, Heartbeat in the classic protocol; ConsumerGroupHeartbeat in the KIP-848 protocol) travel on the connection to the group's coordinator broker.
• Inter-broker replication — follower brokers open a separate connection to each leader they replicate from, used by their replica-fetcher thread. The same fetch API serves both consumers and followers, but inter-broker fetch uses a dedicated replication listener (inter.broker.listener.name) so it can be ACL'd / SASL'd independently from client traffic.
• Controller plane (KRaft) — controllers form a Raft quorum on a dedicated listener (controller.listener.names). Brokers register with the controller quorum and receive metadata updates via BrokerRegistrationRequest / BrokerHeartbeatRequest / FetchRequest against the metadata log. This traffic is small compared to data traffic but is what every cluster decision rides on.
• Zero-copy on the consume hot path — for the response to a FetchRequest, the broker calls FileChannel.transferTo(), which typically maps to sendfile() at the kernel level (the JDK uses the syscall when the OS supports it). Bytes go from page cache straight to the socket buffer — never copied into JVM user space. This is the single biggest reason a broker can saturate a 10 Gbps NIC with a tiny CPU footprint on a tail-following consumer.
• Connection density — Kafka is built for fewer, longer connections rather than many short ones. The network thread pool is sized assuming each socket is mostly idle waiting on the next request, not assuming connection storms; OS-level net.core.somaxconn and fs.file-max tuning matters more on the OS side than JVM tuning.

The hot path for a tail-following consumer never reads from disk — bytes flow page cache → kernel sendfile → NIC with no copy into user space. The hot path for produce is a sequential append into the same page cache, and durability is provided by ISR replication (followers pull through their own fetcher threads), not by fsync per write.

Why Kafka is fast (in one paragraph each)

• Sequential append — every produce is a write to the end of a flat file. SSDs and spinning disks both like this; there are no random seeks, no in-place updates.
• Page cache instead of an internal cache — keeping JVM heap small lets the OS use the rest of RAM as page cache. A tail-following consumer hits the same page the producer just wrote, so there is no disk I/O on the hot path.
• Zero-copy on consume — FileChannel.transferTo() calls sendfile(); bytes go page cache → NIC without traversing user-space buffers. CPU spends no cycles on memcpy.
• Batching + compression — producers batch records per partition before sending; brokers store, replicate, and serve those batches as-is. Compression is end-to-end (decompress only at the consumer).
• Pull-based replication — followers fetch from the leader on their own schedule with the same fetch API as a consumer; the leader has no per-follower send loop. ISR membership is driven by how fresh each follower's pull is.

5. CLI Commands

The scripts ship inside the Kafka distribution's bin/ directory. Examples assume BOOTSTRAP=localhost:9092. For a secured cluster, add --command-config client.properties.

Topic lifecycle — kafka-topics.sh

# Create a topic with 6 partitions and replication factor 3
kafka-topics.sh --bootstrap-server $BOOTSTRAP \
  --create --topic orders \
  --partitions 6 --replication-factor 3 \
  --config min.insync.replicas=2 \
  --config retention.ms=604800000

# List all topics (excluding internal __ topics)
kafka-topics.sh --bootstrap-server $BOOTSTRAP --list \
  --exclude-internal

# Describe a topic: leader, replicas, ISR per partition
kafka-topics.sh --bootstrap-server $BOOTSTRAP --describe --topic orders

# Increase partition count (cannot decrease)
kafka-topics.sh --bootstrap-server $BOOTSTRAP \
  --alter --topic orders --partitions 12

# Delete (requires delete.topic.enable=true on brokers)
kafka-topics.sh --bootstrap-server $BOOTSTRAP --delete --topic orders

Inspect or change topic config — kafka-configs.sh

# Show all configs for a topic (only dynamically overridden ones)
kafka-configs.sh --bootstrap-server $BOOTSTRAP \
  --describe --entity-type topics --entity-name orders

# Add or update a topic-level config
kafka-configs.sh --bootstrap-server $BOOTSTRAP \
  --alter --entity-type topics --entity-name orders \
  --add-config min.insync.replicas=2,retention.ms=604800000

# Remove a previously-overridden config (revert to broker default)
kafka-configs.sh --bootstrap-server $BOOTSTRAP \
  --alter --entity-type topics --entity-name orders \
  --delete-config retention.ms

# Same pattern works for entity-type brokers, clients, users, ips

Quick produce / consume — kafka-console-producer.sh / kafka-console-consumer.sh

# Produce: one record per line, key:value separated by ':'
kafka-console-producer.sh --bootstrap-server $BOOTSTRAP \
  --topic orders \
  --property "parse.key=true" --property "key.separator=:"
# > order-42:{"id":42,"amount":1990}

# Consume from the beginning of the topic, printing key, partition, offset, timestamp
kafka-console-consumer.sh --bootstrap-server $BOOTSTRAP \
  --topic orders \
  --from-beginning \
  --property print.key=true --property print.partition=true \
  --property print.offset=true --property print.timestamp=true

# Consume as part of a consumer group, with manual commit disabled
kafka-console-consumer.sh --bootstrap-server $BOOTSTRAP \
  --topic orders --group billing \
  --consumer-property enable.auto.commit=false

Consumer groups and offsets — kafka-consumer-groups.sh

# List all consumer groups
kafka-consumer-groups.sh --bootstrap-server $BOOTSTRAP --list

# Describe a group: current offset, log-end-offset, lag per partition
kafka-consumer-groups.sh --bootstrap-server $BOOTSTRAP \
  --describe --group billing

# Reset offsets to the earliest available for the entire topic
kafka-consumer-groups.sh --bootstrap-server $BOOTSTRAP \
  --group billing --topic orders \
  --reset-offsets --to-earliest --execute

# Reset to a specific timestamp (epoch ms) — useful for replay windows
kafka-consumer-groups.sh --bootstrap-server $BOOTSTRAP \
  --group billing --topic orders \
  --reset-offsets --to-datetime 2026-06-20T00:00:00.000 --execute

# Delete a consumer group (only when no active members remain)
kafka-consumer-groups.sh --bootstrap-server $BOOTSTRAP \
  --delete --group billing

Partition reassignment — kafka-reassign-partitions.sh

# 1. Generate a candidate plan to move two topics across brokers 1,2,3
cat > topics.json <<'JSON'
{ "version": 1, "topics": [ { "topic": "orders" }, { "topic": "payments" } ] }
JSON

kafka-reassign-partitions.sh --bootstrap-server $BOOTSTRAP \
  --topics-to-move-json-file topics.json \
  --broker-list "1,2,3" --generate

# 2. Save the proposed assignment to reassign.json, then execute
kafka-reassign-partitions.sh --bootstrap-server $BOOTSTRAP \
  --reassignment-json-file reassign.json --execute \
  --throttle 50000000   # bytes/sec replication throttle

# 3. Verify when the move is complete, then drop the throttle
kafka-reassign-partitions.sh --bootstrap-server $BOOTSTRAP \
  --reassignment-json-file reassign.json --verify

ACLs — kafka-acls.sh

# Allow user 'billing-app' to consume topic 'orders' as group 'billing'
kafka-acls.sh --bootstrap-server $BOOTSTRAP \
  --add --allow-principal User:billing-app \
  --operation Read --topic orders \
  --operation Read --group billing

# Allow a producer to write to 'orders' (Write + Describe on the topic)
kafka-acls.sh --bootstrap-server $BOOTSTRAP \
  --add --allow-principal User:order-svc \
  --operation Write --operation Describe --topic orders

# List existing ACLs scoped to a topic
kafka-acls.sh --bootstrap-server $BOOTSTRAP \
  --list --topic orders

Source Tree (for contributors)

Official source: github.com/apache/kafka (read-only GitHub mirror of the Apache Git repository). Apache Kafka is governed by the Apache Software Foundation — changes of any meaningful scope go through a KIP (Kafka Improvement Proposal) on the dev mailing list before code lands. Bug fixes and small improvements go directly as GitHub PRs that a committer reviews and pushes. Build is Gradle (./gradlew jar, ./gradlew test); JDK 17+; the broker (core/) is still Scala, almost everything else is Java.

Vertical layers = dependency depth. Blue arrow = "depends on". Red arrow = "regenerates" (touch a JSON message spec → ./gradlew :generator:run rewrites clients/src/generated/...). A typical broker bugfix lives in core/ or one of the broker-side libs and is verified by a JUnit test in the same module plus a ducktape scenario under tests/. Wire-format changes start in clients/src/main/resources/common/message/*.json, never in the generated Java.

Top-level modules

• clients/ — Java producer, consumer, admin client, and the protocol / serialization layer they share with the broker (org.apache.kafka.common.*). The most-changed module after core/; any change to wire format, partition assignment, or client-side coordinator logic lives here.
• core/ — the broker. Scala. Owns request handling (KafkaApis), replication (ReplicaManager, Partition), the network layer (SocketServer, Processor), and the log subsystem entry points. KRaft refactor moved a lot out of core/ into metadata/ and raft/, but core/ is still the JVM that ships as kafka-server-start.sh.
• server/ + server-common/ — broker-side modules extracted out of core/ as Java. server-common/ is intentionally usable by both clients and broker; server/ is broker-only.
• raft/ — KRaft consensus protocol implementation (the Apache Kafka–native Raft used by the metadata quorum). Independent of ZooKeeper since 3.3.
• metadata/ — KRaft metadata records, log compaction, broker registration, controller state machine. The "what the controller thinks the cluster looks like" layer.
• coordinator-common/, group-coordinator/, transaction-coordinator/, share-coordinator/ — the four coordinator-style subsystems. Group coordinator handles consumer-group rebalancing; transaction coordinator handles EOS transaction state; share coordinator (KIP-932, 4.x) owns shared-consumption / queue semantics. All four were extracted as standalone modules during the KRaft era to make them testable without a full broker.
• storage/ — the log subsystem: LogSegment, index files, log cleaner, tiered storage (KIP-405) interfaces. Performance-sensitive; benchmarked separately via jmh-benchmarks/.
• streams/ — Kafka Streams DSL + Processor API + state stores (RocksDB and in-memory). Lives as its own subproject with its own integration tests and docs.
• connect/ — Kafka Connect runtime, framework, REST API, and the bundled file / mirror-maker connectors.
• generator/ — code-generates protocol message classes from JSON schemas under clients/src/main/resources/common/message/. Touch this if you add a new request/response type — every wire-format change starts with a JSON schema edit + regenerate.
• tools/ — the Java behind the bin/*.sh entry points (kafka-topics.sh, kafka-consumer-groups.sh, kafka-reassign-partitions.sh, ...).
• jmh-benchmarks/ — JMH microbenchmarks. Any perf-sensitive change is expected to come with a benchmark here.
• examples/ — minimal sample producers, consumers, Streams apps. Used in docs.
• tests/ — Python ducktape system tests. Provision real broker clusters in containers / VMs and exercise end-to-end behaviour. Separate from JVM unit / integration tests.
• trogdor/ — fault-injection / chaos framework used by the system tests.
• shell/ — interactive metadata shell (kafka-metadata-shell.sh) for inspecting KRaft snapshots.
• docker/ + docs/ — official Docker image build, and the AsciiDoc/HTML source of kafka.apache.org/documentation.
• bin/ — the shell scripts that ship in the binary distribution; thin wrappers around the Java classes in tools/ and core/.
• checkstyle/, gradle/, licenses/, release/, vagrant/, committer-tools/ — build/style/release machinery. Rarely touched by feature PRs.

Version history

• 0.7 (2012, Apache Incubator) — original Scala broker contributed by LinkedIn. No replication.
• 0.8 (2013) — intra-cluster replication (ISR). The moment Kafka became safe to run as a system of record.
• 0.9 (2015) — new Java consumer (KafkaConsumer) replacing the high-level Scala consumer; security (SSL, SASL, ACLs); Connect framework.
• 0.10 (2016) — Kafka Streams. Record timestamps.
• 0.11 (2017) — idempotent producer + transactions (the foundation of EOS).
• 1.x – 2.x (2017 – 2020) — broker hardening, exactly-once Streams, KIP-500 design discussions (replace ZooKeeper).
• 3.0 (Sep 2021) — KRaft preview (still ZooKeeper-default).
• 3.3 (Oct 2022) — KRaft GA for new clusters; ZooKeeper still supported for upgrades.
• 3.5 – 3.9 (2023 – 2024) — ZooKeeper-to-KRaft migration path stabilises; tiered storage (KIP-405) ships.
• 4.0 (2025) — ZooKeeper removed entirely; KRaft is the only metadata mode. New consumer rebalance protocol (KIP-848) becomes the default.
• 4.1 – 4.3 (current line) — ongoing development of share groups (KIP-932) queue-style consumption, further coordinator splits, ongoing tiered-storage polish.

Branching & release model

trunk (the repository's main branch is named trunk) is the next minor. Each released minor gets a long-lived branch (4.0, 4.1, 4.2, 4.3); bugfixes are committed to trunk first then cherry-picked back to every still-supported branch by the committer. Release cadence is roughly one minor every 4 months. The Apache release process is run by the release manager: tag X.Y.Z-rcN, vote on dev@kafka.apache.org, promote to X.Y.Z. Tags published to GitHub follow the same names; the actual .tar.gz artefacts live on downloads.apache.org.

Related

• Speckafka.apache.org/documentation
• My PRsApache Kafka contributions
• RelatedSpring Kafka — silent record loss (GH-4504)
• Blogdevbilllab.tistory.com