Why Apache Kafka doesn't need fsync to be safe — Jack Vanlightly
April 24, 2023
Why Apache Kafka doesn't need fsync to be safe
Image credit: G. Hüdepohl (atacamaphoto.com)/ESO
First the TLDR!
TLDR: Apache Kafka doesn’t need fsyncs to be safe because it includes recovery in its replication protocol. It is a real-world distributed system that uses asynchronous log writing + recovery with some additional extra safety built-in. Asynchronous log writing allows it to provide robust performance on a variety of hardware and with a wide variety of workloads.
Now that the TLDR is done, let’s dive into it.
The fact that by default Apache Kafka doesn’t flush writes to disk is sometimes used as ammunition against it. The argument is that if Kafka doesn’t flush data before acknowledging produce requests then surely the cluster can lose acknowledged data due to crashes and reboots. It sounds plausible and so people may believe it - but I’m here writing this today to explain why that isn’t the case.
What is an fsync / flush?
The term "fsync" and “flush” refers to an operating system call (known as a sys call) that ensures that data written to a file is physically stored on the storage drive. It is used to synchronize a file's in-memory state with its on-disk state.
Basically, if data is written but no fsync is performed and then the computer itself were to crash, be rebooted or lose power, those file changes would not be persisted.
Kafka does not flush synchronously - that means that, by default, it will not always flush before acknowledging a message. Before we explain why async flushes are ok for Apache Kafka, let’s look at why they’re unsafe for Raft.
Why Raft needs fsyncs
I am not picking on Raft - I love Raft! Just check out my GitHub repo on Raft, and blog post about Flexible Raft. Go check out RabbitMQ Quorum Queues which I was lucky enough to be a part of while at VMware. Just look at KRaft in Apache Kafka! Raft is awesome, and it needs fsyncs.
In a nutshell
The Raft protocol is essentially a group of servers where each server runs the same logic and is backed up by a log of operations persisted to disk. This protocol includes leader elections, failure detection, data replication, changing cluster membership and so on.
View fullsize
Fig 1. Raft is a state-machine replication (SMR) algorithm which consists of a cluster of state machines which sit on a replicated log.
A Raft cluster cannot tolerate losing any log entries that were previously written to disk. Raft depends on the log itself for the correctness of the protocol and so if the log on one node loses data it can compromise the correctness of the entire cluster.
If an entry is written to disk but then the server crashes and loses that entry, Raft can lose the copies of that data stored on other servers - that is, losing an entry on one server can cause all servers to lose that entry! Therefore each Raft server must fsync data written to disk.
The why (for those that want to understand more)
The main reason for the dependency on the integrity of the Raft log is that leader elections can cause servers to delete acknowledged data if any Raft server loses entries in its log.
Raft requires majority quorums to operate. For example, if we have three nodes, then there are three possible quorums that form a majority (of 2 nodes).
View fullsize
Fig 2. Three possible quorums of two nodes: {n1,n2}, {n2,n3} and {n1,n3}.
There are two types of quorum in Raft:
-
An election quorum: a group of nodes that all vote for the same node.
-
A replication quorum: a group of nodes that host a given message (and all prior messages).
A successful election requires the following things:
-
A majority of nodes to vote for the same node.
-
Each node must only vote for another if it has the same or more recent data. A node without the latest data won’t be elected.
The crux of the design is that Raft guarantees that the replication quorum of the most recent entries overlaps with all possible election quorums. That is to say, at least one node of any election quorum will have the latest data and only the node with the most recent data can win the election - so we guarantee that the new leader has all the data.
The overlapping of replication and election quorums is the cornerstone of Raft.
Fig 3. There exist no two replication and election quorums that do not overlap.
So what happens to Raft if nodes don’t fsync?
The answer is that the cornerstone of overlapping quorums is broken.
A simple example of Raft data loss without fsync
Let’s use a simple example of a Raft cluster with a handful of messages.
Messages m1, m2 and m3 have been added to the log, replicated to by a majority and acknowledged to the clients. Raft guarantees that these three won’t be lost as long as a majority of nodes are not lost.
View fullsize
Raft example step1: all is good.
But then the server of broker 1 fails/loses power, and it hadn’t performed an fsync causing m2 and m3 to be lost.
View fullsize
Raft example step2: broker 1 loses m2 and m3.
Broker 1 comes back with only m1. Broker 1 starts an election (ignoring pre vote here).
Raft example step3: Broker 1 is back with only m1.
Now b1 and b3 can form an election quorum which does not overlap with the replication quorum of m2 and m3 (which is only b2 now).
In the end, b1 and b3 both vote for b1, and so b1 becomes the leader. This causes b2 tol delete m2 and m3 from its log.
View fullsize
Raft example step 4: Oh no! An election and replication quorum that don’t overlap!
The Raft cluster just lost two acknowledged messages.
View fullsize
Raft example step 5: Broker 2 truncates its log to m1. The cluster just lost two acknowledged entries.
So vanilla Raft needs fsyncs to be safe and all implementations I know of do so. Of course you could modify Raft to not fsync but you’d need a recovery protocol and changes in when a node can be a candidate for leader. In fact, that is exactly what we’ll look at now, but with the Kafka replication protocol.
Why is Kafka safe despite not flushing every message?
In a nutshell
The answer is that Kafka’s data replication protocol was designed to be safe without fsyncs. Kafka replication does not depend on the integrity of a topic partition for leader elections, instead it depends on an external metadata store such as ZooKeeper or KRaft. A topic partition can lose some messages without compromising the correctness of the data replication protocol itself.
In order to cope with the loss of some messages on one broker, Kafka has in-built recovery or self-healing mechanism that automatically repairs the data. This recovery mechanism is what makes Kafka safe without fsyncs.
The why (for those that want to understand more)
Asynchronous log writing has been the subject of much research in the distributed systems community, with many real-world examples such as VSR Revisited ( 1, 2, 3) and Apache BookKeeper with the journal disabled.
There is one down-side to asynchronous log writing with these two systems - simultaneous crashes can cause data loss. Having run different distributed systems in production, in some cases at very large scale, I have seen cluster crashes occur, such as cascading OOMs taking out a cluster under high load, or a bug causing all nodes to crash at about the same time. Kafka is not subject to that issue because it doesn’t store unflushed data in its own memory, but in the page cache. If Kafka crashes, the data doesn’t go bye bye and for this reason Kafka is actually safer than BookKeeper without the journal - despite both using recovery protocols.
So how does the Kafka data replication protocol get away without fsyncs? It’s all about recovery.
Kafka replication and recovery
All distributed systems that write to storage asynchronously have a recovery mechanism built-in that allows an affected node to recover any data that was lost due to an abrupt termination of the node.
All these recovery mechanisms have the following rules in common:
-
The affected node must recover lost data from a peer.
-
The affected node cannot be a candidate for leadership until it has at least recovered the data it previously lost.
The recovery process with Kafka is very simple.
-
A failed broker restarts, having lost some number of messages that were not successfully stored to disk. Let’s say the log had 1000 messages, but lost the last ten meaning the last offset is now 990.
-
The broker has been removed from the ISR (the quorum which are leader candidates). The broker starts fetching at offset 991 from the existing topic partition leader, adding the messages to its own log. The first ten messages recover those lost earlier.
-
Once the broker is caught up, it is able to join the ISR as a fully functional broker that is a candidate for leadership.
The point is that until the affected broker catches up (and recovers any lost data), it cannot become a leader and therefore there can be no data loss. This concept of a node not being a leader candidate until it has at least recovered what it lost is the fundamental principle behind all three recovery protocols I am aware of.
An example of recovery
View fullsize
Kafka example step 1: All is good though the ISR is down to two to make this similar to the “Raft without fsync” example.
For whatever reason, let’s say the ISR only includes a majority, like our Raft example. This means that acknowledged messages might only exist on 2 brokers, as is the case with messages m1, m2 and m3.
Then the server of b1 crashes/loses power, causing m2 and m3 to be lost on broker 1. At this point broker 1 is removed from the ISR and stops being a candidate for leadership.
View fullsize
Kafka example 2: The server of broker 1 goes offline, taking the page cache with it and losing m2 and m3.
Broker 1 comes back online, without messages m2 and m3.
View fullsize
Kafka example step 3: Broker 1 comes back without the unflushed messages m2 and m3.
Broker 1 starts fetching messages from the leader, starting at offset of m1 + 1. Once broker 1 has caught up, it gets included in the ISR again. Any future leader elections can safely include broker 1.
View fullsize
Kafka example step 4: Broker 1 fetches from broker 2 until it is caught up and rejoins the ISR.
Only brokers in the ISR can become the leader and all brokers in the ISR have the complete log - there is no equivalent data loss scenario like with “Raft without fsync”.
The Last Replica Standing issue
There is however one edge case where Kafka, as of version 3.5 today, does not correctly handle recovery which can allow for data loss. I have written about this edge case and KIP-966 which addresses it.
Handling cluster-wide failures
Kafka may handle simultaneous broker crashes but simultaneous power failure is a problem. If all brokers lose power simultaneously then there is a risk of data loss. Every system makes trade-offs and this is one of Apache Kafka’s. However, luckily for us, there’s a simple solution.
High availability and high durability are only truly possible when clusters are spread over multiple data centers. In the cloud this means spreading clusters over multiple availability zones where each zone is actually a physically separated data center. This avoids disasters such as fires or power surges from destroying all copies of the data and allows a system to remain available.
Asynchronous log writing designs like Kafka are most durable when spread over multiple zones to avoid power outages from causing all brokers to simultaneously and abruptly stop which could lead to some unrecoverable data. This is a common argument against Apache Kafka however the truth is that organizations that care about durability should be using multi-zone deployments in any case.
In summary
Replication protocols generally need for each member to act monotonically, that is, if we ask Node 1 if it has an entry now and it say “yes!” then we assume it will always say yes. If the same broker later says, “sorry, what entry?” then a leader election can result in data loss. But with a recovery protocol, we can tolerate nodes losing data. With Kafka, as long as a single broker in the ISR retains the committed data, then the recovery protocol can do its work.
If you are interested in the details of how we strengthened the recovery part of the Kafka Replication Protocol, then check out my protocol description and TLA+ specification.
Safety isn’t binary, there is no one single thing to guard against data loss. Disk failures, fsync not working as expected, data corruption, bugs! Kafka's recovery protocol aims to balance high performance on all kinds of disks with safety. Correlated failure from a simultaneous power loss event is a risk, but one mostly mitigated by multi-AZ deployments. Kafka offers a dial for fsyncs for those who want the added safety of withstanding correlated failures.
Comments (10)
Newest FirstOldest FirstNewest FirstMost LikedLeast Liked
Preview Post Comment…
reuven3 years agoPendingAwaiting Moderation· 0 Likes
There is a middle ground for better resilience without a multi-zonal deployment - battery backup. If one has servers with battery backup, normal power outages won't cause correlated failures across brokers as the page cache will still flush on power outage. This still leaves one vulnerable to more extreme scenarios (fires, lightning strikes, meteor strikes, etc.) which do require a multi-zonal solution.
Preview Post Reply
Nicholas3 years agoPendingAwaiting Moderation· 0 Likes
The article somewhat glosses over one little detail: what is the sequence for the returning node to determine that it has lost messages and thus not ISR eligible?
Preview Post Reply
Jack Vanlightly3 years agoPendingAwaiting Moderation· 0 Likes
In KIP-966, the broker writes its broker epoch to file on shutdown. If it was an unclean shutdown, no epoch is written. On boot, the broker registers with the controller and passes its persisted broker epoch. If there is no broker epoch on disk it sends an empty value. The controller now knows this broker is unclean and removes it from any ISRs/ELRs.
Preview Post Reply
Igor3 years agoPendingAwaiting Moderation· 0 Likes
Thanks you for great post.
Do you know what happened in this scenario, why redpanda claiming that kafka lost message ? https://redpanda.com/blog/why-fsync-is-needed-for-data-safety-in-kafka-or-non-byzantine-protocols
Preview Post Reply
Jack Vanlightly3 years agoPendingAwaiting Moderation· 0 Likes
Yes this is a protocol bug that was identified, called the Last Replica Standing issue. It is being addressed in KIP-966. You can see my write-up of that here: https://jack-vanlightly.com/blog/2023/8/17/kafka-kip-966-fixing-the-last-replica-standing-issue
Basically this is a protocol bug. Unfortunately bugs happen and project maintainers fix them, this is one such case. I always say that safety is not a binary value and if you search the release notes of Redpanda, Kafka, Pulsar or any other distributed data system you will find data loss bugs getting fixed (even recently). For Redpanda to claim push-button safety, considering the complexity of its implementation, is just marketing. The key is that these projects work hard to minimize the likelihood of these bugs happening but when they eventually do happen, fixing them as soon as they can.
I know my former colleagues in the RabbitMQ team use extensive fault-injection testing and stress testing. Most of the core protocols are verified in TLA+ (by myself). Likewise, Pulsar/BookKeeper has similar testing and I formally verified BookKeeper with TLA+, detecting and fixing a bug in the protocol. The Kafka community also use these techniques and I have been busy formally verifying its protocols (KRaft, partition replication, consumer group protocol). Even so, bugs can still happen, but we fix them when they do happen.
Preview Post Reply
Tony3 years agoPendingAwaiting Moderation· 0 Likes
Thank you for the clear exposition and good use of illustrations. The reference to high availability reminded me of potential failure modes related to network partitions, which become more probable once the cluster is spread out across multiple availability zones. Perhaps consider writing a follow-on article discussing the interplay between Kafka client configurations (acks and retries, etc.) and the various cluster failure modes.
Preview Post Reply
Jack Vanlightly3 years agoPendingAwaiting Moderation· 0 Likes
This may interest you: https://github.com/Vanlightly/kafka-tlaplus/blob/main/kafka_data_replication/kraft/kip-966/description/2_replication_algorithm.md#25-produce-request-durability-level-acks
It's worth reading the whole protocol description though: https://github.com/Vanlightly/kafka-tlaplus/blob/main/kafka_data_replication/kraft/kip-966/description/0_kafka_replication_protocol.md
Preview Post Reply
Luca3 years agoPendingAwaiting Moderation· 0 Likes
ArcadeDB uses the same mechanism for election: a node that is not updated to the last message can't be voted as a leader. I thought this was standard RAFT, but seems it's not and it's an easy variant to avoid requiring the fsync. Nice article, well explained!
Preview Post Reply
Jack Vanlightly3 years agoPendingAwaiting Moderation· 0 Likes
With Raft the rule is that a node cannot vote for a peer if the peer has a lower last index/term. This combined with a majority vote ensures safety, as long as no entries were ever lost.
Preview Post Reply
Alan3 years agoPendingAwaiting Moderation· 0 Likes
Thanks for the great explanation. I didn't know that Raft needed to fsync to rely on consistency
Preview Post Reply