Kafka: a Distributed Messaging System for Log Processing

In one breath — the whole paper, compressed

LinkedIn's activity and operational log data was orders of magnitude larger than its "real" data, but enterprise messaging systems were too feature-heavy and too slow for it, while log aggregators such as Scribe and Flume served only offline batch consumers. Kafka splits each topic into partitions spread over a cluster of brokers and stores each partition as nothing more than an append-only sequence of roughly 1GB segment files, using a message's logical offset in the log as its only identifier. Consumers pull rather than being pushed to: each request names a starting offset and a byte budget, the broker locates the segment through a small in-memory offset index and ships bytes straight from file to socket with sendfile, and the broker keeps no per-consumer delivery state at all. Consumption position lives in ZooKeeper alongside broker, consumer and partition-ownership registries, so consumer groups rebalance themselves with no master, and messages are retained on a time-based SLA (typically 7 days) rather than deleted on acknowledgement, which lets any consumer rewind and replay. The measured result was 50,000 messages per second unbatched and 400,000 batched from a single producer against one broker - at least twice RabbitMQ and orders of magnitude past ActiveMQ - with at-least-once delivery as the only guarantee offered.

Before this paper — the world it landed in

By 2011 consumer internet companies were producing log data - pageviews, clicks, searches, likes, plus service call latencies and per-machine CPU, memory, network and disk metrics - at volumes orders of magnitude beyond their transactional data; the paper cites China Mobile collecting 5-8TB of phone call records and Facebook gathering almost 6TB of user activity events every day. What had changed was not the volume but the consumer: this data was moving out of nightly analytics and into live site features - search relevance, recommendations, ad targeting and reporting, spam and scraping defence, newsfeed aggregation - that needed it within a few seconds. Neither available toolkit fit. Enterprise messaging systems such as IBM Websphere MQ, JMS implementations, TIBCO EMS and Oracle EMS offered transactional inserts across multiple queues and per-message out-of-order acknowledgement, guarantees that are overkill when losing the occasional pageview is harmless; they had no producer batching API so every message cost a full TCP/IP roundtrip, weak support for partitioning across machines, and sharply degraded performance once unconsumed messages were allowed to accumulate. Specialized log aggregators - Facebook's Scribe, Yahoo's Data Highway, Cloudera's Flume - scaled fine but were built to dump into HDFS or NFS for offline consumption, leaked implementation details such as "minute files" to their consumers, and used a push model that floods a slow consumer instead of letting it pull at its own sustainable rate.

The problem — what was actually breaking

Enterprise messaging systems spend their complexity budget on delivery guarantees such as atomic inserts into multiple queues and per-message out-of-order acknowledgement, which are overkill for log data where losing a few pageview events occasionally is not the end of the world.
JMS provides no API for a producer to batch multiple messages into a single request, so each message requires a full TCP/IP roundtrip, which is not feasible at log-processing throughput.
Existing messaging systems are weak in distributed support, offering no easy way to partition and store messages across multiple machines.
Messaging systems assume near-immediate consumption and degrade significantly when unconsumed messages accumulate, which is exactly the situation created by offline warehouse consumers doing periodic large loads instead of continuous consumption.
Existing log aggregators are built for offline consumption, expose implementation details such as "minute files" to consumers, and cannot serve the real-time applications LinkedIn needed with delays of no more than a few seconds.
Log aggregators use a push model in which the broker forwards data to consumers, so a consumer can be flooded faster than it can handle and cannot easily be rewound to re-read older data.

Core ideas — the contributions, and why they work

The topic as a partitioned log

A topic is a named stream of messages: producers publish to it, brokers store it, consumers subscribe to it. Kafka's structural move is to divide each topic into partitions and scatter those partitions across the brokers of a cluster, so a topic's throughput is not bounded by one machine's disks or network card. Each partition is a logical log with a total order, and the partition - not the message, not the topic - is the smallest unit of parallelism: at any given time all messages from one partition are consumed by exactly one consumer within each group. That single decision buys ordering within a partition and removes any need for locking or per-message state maintenance, turning scale-out into a matter of over-partitioning the topic.

Offsets instead of message ids

A message stored in Kafka has no explicit message id; it is addressed by its logical offset in the log, and the id of the next message is the current id plus the current message's length. Ids therefore increase but are not consecutive, and the design eliminates the auxiliary, seek-intensive random-access index structures that conventional brokers maintain to map message ids to storage locations. The broker keeps only an in-memory sorted list of offsets, including the offset of the first message in every segment file, which is enough to find the right file and then read forward sequentially. Because a consumer always reads a partition in order, acknowledging one offset implies receipt of everything before it, collapsing per-message delivery state into a single number.

Pull, not push

Consumers issue asynchronous pull requests carrying a starting offset and an acceptable number of bytes, rather than having the broker push messages at them. Each consumer therefore retrieves at the maximum rate it can sustain and can never be flooded by a producer or by a broker draining a backlog - the reason LinkedIn preferred pull over the push model used by Scribe, Data Highway and Flume. Because position is a parameter of the request rather than server-side state, a consumer can also move it backwards, and the paper treats rewinding as essential rather than incidental. It is how a consumer replays messages after a bug in its own application logic is fixed, and how a full-text indexer that flushes only periodically restarts from the smallest unflushed offset after a crash.

Stateless broker with time-based retention

Unlike most messaging systems, a Kafka broker does not track how much each consumer has consumed; that information belongs to the consumer. This removes a great deal of bookkeeping and disk write traffic from the broker, but it also means the broker cannot know when a message is safe to delete. Kafka answers with a deliberately blunt policy: a message is deleted once it has been retained longer than a configured period, typically 7 days, regardless of who read it. This works because Kafka's performance does not degrade as the log grows, so a week of retention costs only disk, and because in practice consumers - including the offline ones - finish daily, hourly, or in real time.

No application cache: page cache plus sendfile

Kafka deliberately avoids caching messages in its own process memory, relying instead on the underlying file system page cache. This avoids double buffering, keeps the cache warm even when a broker process is restarted, and leaves the JVM with almost nothing to garbage-collect, which is what makes an efficient implementation in a VM-based language feasible. Because producers append sequentially and consumers usually lag the producer by only a small amount, the ordinary operating system heuristics - write-through caching and read-ahead - happen to be exactly the right ones, and the paper reports production and consumption performance linear in data size up to many terabytes. On the read path Kafka then uses the Unix sendfile API to move bytes from a log segment's file channel directly into a socket channel, a win multiplied by the fact that Kafka is multi-subscriber and the same bytes are shipped repeatedly.

Message sets on both ends

Batching is a first-class part of the API, not an afterthought. A producer sends a MessageSet in a single publish request; a consumer's pull request likewise returns many messages, typically hundreds of kilobytes' worth, even though the client-side iterator hands them to the application one at a time. This amortizes the RPC and TCP/IP roundtrip cost that the paper identifies as the fatal flaw of JMS-style per-message publishing, and it is also what makes the sendfile path worthwhile, since a large contiguous run of bytes is being shipped in one call. In the experiments, raising the batch size from 1 to 50 improved producer throughput by almost an order of magnitude.

Decentralized coordination, no master node

Rather than electing a master broker or a central coordinator, Kafka lets consumers coordinate among themselves in a decentralized fashion, on the explicit argument that a master is one more failure mode to worry about. ZooKeeper supplies the substrate: a broker registry, a consumer registry, a per-group partition ownership registry, and a per-group offset registry, of which the first three are ephemeral paths that disappear automatically when their creator dies. Each consumer registers watchers on the broker and consumer registries, so any membership change notifies everyone, and every consumer then independently runs the same deterministic assignment over the same sorted inputs. Coordination therefore happens only at rebalance time, an infrequent event, and never on the message path.

How it works — the mechanism, concretely

Producing: partition selection and the publish request

A message is defined to contain just a payload of bytes, so the user chooses her own serialization; LinkedIn layered Avro on top, storing the id of the Avro schema plus the serialized bytes in the payload and resolving ids through a lightweight schema registry service. Messages are gathered into a MessageSet and handed to a single send call naming the topic. The producer picks the target partition either at random or by applying a partitioning function to a partitioning key - the hook that makes all messages sharing a join key land on one partition and therefore in one consumer process. The producer does not wait for broker acknowledgements and sends as fast as the broker can handle, which is what pushes publish throughput up but also means an unacknowledged message can silently be lost.

Broker storage: segment files and the in-memory offset index

Each partition of a topic corresponds to a logical log, physically implemented as a set of segment files of approximately the same size, for example 1GB. Publishing is simply an append to the last segment file - no seek, no index update, no per-message metadata write. Segment files are flushed to disk only after a configurable number of messages have been published or a configurable interval has elapsed, and a message is exposed to consumers only after it has been flushed, which is where durability is actually decided. The broker keeps in memory a sorted list of offsets including the offset of the first message in every segment file, so serving a fetch means searching that list to find the containing segment and then reading forward; retention deletes whole segments from the front of the log.

Consuming: message streams, pull requests and offset arithmetic

A consumer calls createMessageStreams for a topic and receives one or more message streams over which the published messages are evenly distributed; each stream exposes an iterator that, unlike an ordinary iterator, never terminates - it blocks when the log is exhausted and resumes when new messages arrive. Underneath, the consumer issues asynchronous pull requests, each carrying the offset at which consumption begins and an acceptable number of bytes to fetch, keeping a buffer of data ready for the application. After receiving a message the consumer computes the next offset by adding the message's length and uses it in the following request. Point-to-point delivery falls out of putting several consumers in one group; publish/subscribe falls out of putting them in different groups, which need no coordination with one another at all.

Zero-copy delivery with sendfile

The conventional path from a local file to a remote socket is four data copies and two system calls: storage media to OS page cache, page cache to an application buffer, application buffer to a kernel socket buffer, and socket buffer out to the network. Kafka instead calls the Unix sendfile API to transfer bytes directly from the log segment's file channel to the socket channel, avoiding two of the copies and one of the system calls. This is only possible because the on-disk representation and the on-wire representation are the same bytes: the broker does no per-consumer transformation, no re-framing and no id-to-location lookup. The payoff compounds because Kafka is a multi-subscriber system in which a single message may be consumed many times by different applications.

ZooKeeper registries and the rebalance algorithm

On startup a broker writes its host name, port and its set of topics and partitions into the broker registry, and a consumer writes its group and subscribed topics into the consumer registry. Each consumer group additionally owns an ownership registry - one path per subscribed partition whose value is the id of the consumer currently reading it - and an offset registry recording the last consumed offset per partition. Broker, consumer and ownership paths are ephemeral, so a failure automatically removes the failed party's entries; only the offset registry is persistent. When a watcher fires, consumer Ci runs Algorithm 1: remove its partitions from the ownership registry, read both registries, compute the available partition set PT and the subscribing consumer set CT for topic T, sort both, let j be its index in CT and N = |PT|/|CT|, claim partitions j*N through (j+1)*N-1, write itself as owner, and start one pull thread per claimed partition beginning at the offset stored in the offset registry.

Conflicts, corruption and delivery guarantees

Because rebalance notifications reach consumers at slightly different times, a consumer may try to take ownership of a partition still owned by another; when this happens it releases every partition it owns, waits a bit and retries, which in practice stabilizes after only a few attempts. A brand new consumer group with no stored offsets starts from either the smallest or the largest available offset per configuration, using an API the brokers expose for exactly this. Kafka stores a CRC for each message in the log so that a broker hitting an I/O error can run a recovery process removing messages with inconsistent CRCs, and so that clients can detect network errors after producing or consuming. The delivery guarantee is at-least-once: a consumer that crashes without a clean shutdown leaves messages past its last offset committed to ZooKeeper, and whichever consumer takes over that partition may re-deliver them, so applications that care must de-duplicate using the returned offsets or a unique key in the message.

The LinkedIn deployment: mirroring, auditing and Hadoop loads

Every datacenter running user-facing services hosts a co-located Kafka cluster; frontend services publish log data to it in batches through a hardware load balancer that spreads publish requests evenly over the brokers, and online consumers run in the same datacenter. A separate analysis datacenter, placed close to the Hadoop cluster and warehouse infrastructure, runs a Kafka cluster whose embedded consumers pull from every live cluster, producing a replica against which load jobs, reporting and ad hoc scripts run. Correctness is audited rather than assumed: each message carries a generation timestamp and server name, each producer periodically publishes a monitoring event to a separate topic recording how many messages it sent per topic in a fixed window, and consumers reconcile their received counts against those events. Hadoop ingestion uses a custom Kafka input format so MapReduce jobs read directly from brokers, and because offsets are held client-side, both data and offsets are written to HDFS only on successful completion of the job, so a failed and restarted task neither duplicates nor loses data.

What the paper showed — measurements and proofs

In the producer test - one producer machine and one broker machine, each with 8 2GHz cores, 16GB of memory and 6 disks in RAID 10, connected by a 1Gb link - Kafka published 10 million 200-byte messages at an average of 50,000 messages per second with batch size 1 and 400,000 with batch size 50, orders of magnitude above ActiveMQ v5.4 and at least twice RabbitMQ v2.4.
With a batch size of 50 a single Kafka producer almost saturated the 1Gb link between producer and broker, and batching alone improved throughput by almost an order of magnitude by amortizing the RPC overhead.
In the consumer test, with all systems configured to prefetch roughly the same amount per request (up to 1000 messages or about 200KB) and all data resident in cache, Kafka consumed an average of 22,000 messages per second, more than four times that of ActiveMQ and RabbitMQ.
Kafka's per-message storage overhead averaged 9 bytes against ActiveMQ's 144, meaning ActiveMQ used 70% more space for the same 10 million messages; the authors traced part of that to the heavy JMS message header and observed one of ActiveMQ's busiest threads spending most of its time in a B-Tree maintaining message metadata and state.
During the consumer test there were no disk write activities on the Kafka broker at all, while an ActiveMQ thread was busy writing KahaDB pages to disk - a direct measurement of what maintaining per-message delivery state on the broker costs.
In production at LinkedIn, Kafka accumulated hundreds of gigabytes and close to a billion messages per day across live and analysis datacenters, with an average end-to-end pipeline latency of about 10 seconds achieved without much tuning.

Limits and trade-offs — conceded and discovered

Conceded by the paper: Kafka has no replication. If a broker goes down, any message stored on it and not yet consumed becomes unavailable, and if the broker's storage is permanently damaged those messages are lost forever; built-in replication across brokers, in both asynchronous and synchronous flavours, is listed as the first item of future work.
Conceded by the paper: the producer does not wait for acknowledgements, so there is no guarantee that a published message was actually received by the broker. The authors accept this explicitly - for many types of log data it is desirable to trade durability for throughput as long as the number of dropped messages is relatively small - while noting they plan to address durability for more critical data.
Conceded by the paper: delivery is only at-least-once. A consumer that crashes without a clean shutdown lets its successor re-deliver messages after the last offset committed to ZooKeeper, and the paper pushes de-duplication onto the application on the argument that this is more cost-effective than two-phase commit. Later Kafka undercut that premise: idempotent producers and transactions in 0.11 delivered exactly-once semantics without two-phase commit on the message path.
Conceded by the paper: ordering is guaranteed only within a single partition, never across partitions, and since a partition is the smallest unit of parallelism a consumer group can never usefully have more consumers than partitions. The prescribed workaround, over-partitioning a topic, pushes a capacity decision forward to topic creation time.
Exposed by later work: client-side coordination through ZooKeeper looked cheap at this scale but did not survive growth - every consumer watching every registry produced herd effects and repeated rebalance storms, and per-partition offset commits made ZooKeeper a write hotspot. Kafka 0.8.2 moved offsets into an internal __consumer_offsets topic, 0.9 moved group membership to a broker-side coordinator, and KIP-500 eventually removed ZooKeeper from Kafka entirely in favour of a Raft-based internal metadata log.

What it became — the systems that inherited it

Kafka became the default event backbone of the industry, and nearly everything the paper describes is still visible in it: topics, partitions, offsets, consumer groups, segment files and time-based retention. The items listed as future work became its most consequential features - replication with in-sync replica sets arrived in 0.8, exactly-once semantics through idempotent producers and transactions in 0.11, and the hoped-for "library of helpful stream utilities" became Kafka Streams and, in the wider ecosystem, Samza, Storm, Spark Streaming and Apache Flink, all of which treat Kafka partitions as their source of truth. The stateless-broker-plus-client-offset design turned out to be the enabling idea for stream processing generally: because the log is replayable and the position is a number the consumer owns, reprocessing history and bootstrapping a brand new consumer are the same operation, which is the basis of the Kappa architecture and of log-based change data capture through Kafka Connect and Debezium. Treating the distributed log as a primitive rather than a queue was picked up directly by AWS Kinesis, Apache Pulsar, Azure Event Hubs, Redpanda and NATS JetStream, and indirectly by database designs that make the replication log the system of record. The efficiency recipe - append-only segments, no application-level cache, sendfile, batched message sets, on-disk format equals on-wire format - became standard practice for high-throughput storage systems well beyond messaging. Kafka's own trajectory closed the loop when KRaft replaced ZooKeeper with an internal Raft metadata log, removing the one external dependency this paper had leaned on.

In the paper’s words — verbatim

“Unlike typical messaging systems, a message stored in Kafka doesn't have an explicit message id. Instead, each message is addressed by its logical offset in the log.”

§3.1

“Unlike most other messaging systems, in Kafka, the information about how much each consumer has consumed is not maintained by the broker, but by the consumer itself.”

§3.1 Stateless broker

“In general, Kafka only guarantees at-least-once delivery. Exactly-once delivery typically requires two-phase commits and is not necessary for our applications.”

§3.3

Vocabulary — as this paper uses it

Topic: A stream of messages of a particular type: producers publish to a topic and consumers subscribe to one or more topics. A topic is divided into partitions that are distributed across the brokers of a cluster.
Partition: One slice of a topic, stored on a broker as a single logical log with a total order. It is Kafka's smallest unit of parallelism - at any time all messages from one partition are consumed by exactly one consumer within each consumer group.
Broker: A server that stores published messages; a Kafka cluster is many brokers, each holding one or more partitions. In this paper a broker keeps no per-consumer state and no replicas of another broker's data.
Offset: The logical position of a message within its partition's log, used in place of an explicit message id. Offsets increase but are not consecutive: the next offset is the current one plus the length of the current message.
Segment file: The physical unit of a partition's log, a file of approximately fixed size such as 1GB. Publishing appends to the last segment, retention deletes whole segments from the front, and the broker's in-memory index records the first offset of every segment.
Message set: A batch of messages carried in one publish request or returned in one pull response, typically hundreds of kilobytes on the consumer side. Batching amortizes the RPC and TCP/IP roundtrip cost that JMS-style per-message publishing cannot avoid.
Consumer group: One or more consumers that jointly consume a set of subscribed topics, with each message delivered to only one member of the group. Different groups each independently consume the full stream and need no coordination between them, which is how one topic serves both point-to-point and publish/subscribe.
Rebalance: The decentralized reassignment of partitions to consumers, triggered by a ZooKeeper watcher when brokers or consumers appear or disappear. Every consumer independently sorts the partition and consumer sets, claims a contiguous range, and resumes from the offset held in the offset registry.
At-least-once delivery: Kafka's only stated guarantee: every message reaches each consumer group at least once, but an unclean consumer crash can cause duplicates after the last offset committed to ZooKeeper. Applications that care must de-duplicate using offsets or a unique key, which the authors argue is cheaper than two-phase commit.

On the timeline — where this sits in the story

Year2011

MilestoneApache Kafka

Problem solvedCompanies needed reliable high-volume event ingestion and a way to decouple producers from consumers.

How it solved itKafka provided a durable distributed commit log with topics, partitions, replay, and consumer groups.

Why it matteredBecame core infrastructure for streaming pipelines, event-driven systems, and real-time data platforms.

DDIACh. 12: Stream Processing

View on the timeline