.svg)
Many Kafka users love the ability to quickly dump a lot of records into a Kafka topic and are happy with the fundamental Kafka guarantee that Kafka is durable. Once a producer has received an ACK after producing a record, Kafka has safely made the record durable and reserved an offset for it. After this, all consumers will see this record when they have reached this offset in the log. If any consumer reads the topic from the beginning, each time they reach this offset in the log they will read that exact same record.
In practice, when a consumer restarts, they almost never start reading the log from the beginning. Instead, Kafka has a feature called “consumer groups” where each consumer group periodically “commits” the next offset that they need to process (i.e., the last correctly processed offset + 1), for each partition. When a consumer restarts, they read the latest committed offset for a given topic-partition (within their “group”) and start reading from that offset instead of the beginning of the log. This is how Kafka consumers track their progress within the log so that they don’t have to reprocess every record when they restart.
This means that it is easy to write an application that reads each record at least once: it commits its offsets periodically to not have to start from the beginning of each partition each time, and when the application restarts, it starts from the latest offset it has committed. If your application crashes while processing records, it will start from the latest committed offsets, which are just a bit before the records that the application was processing when it crashed. That means that some records may be processed more than once (hence the at least once terminology) but we will never miss a record.
This is sufficient for many Kafka users, but imagine a workload that receives a stream of clicks and wants to store the number of clicks per user per hour in another Kafka topic. It will read many records from the source topic, compute the count, write it to the destination topic and then commit in the source topic that it has successfully processed those records. This is fine most of the time, but what happens if the process crashes right after it has written the count to the destination topic, but before it could commit the corresponding offsets in the source topic? The process will restart, ask Kafka what the latest committed offset was, and it will read records that have already been processed, records whose count has already been written in the destination topic. The application will double-count those clicks.
Unfortunately, committing the offsets in the source topic before writing the count is also not a good solution: if the process crashes after it has managed to commit these offsets but before it has produced the count in the destination topic, we will forget these clicks altogether. The problem is that we would like to commit the offsets and the count in the destination topic as a single, atomic operation.
And this is exactly what Kafka transactions allow.
At a very high level, the transaction protocol in Kafka makes it possible to atomically produce records to multiple different topic-partitions and commit offsets to a consumer group at the same time.
Let us take an example that’s simpler than the one in the introduction. It’s less realistic, but also easier to understand because we’ll process the records one at a time.
Imagine your application reads records from a topic t1, processes the records, and writes its output to one of two output topics: t2 or t3. Each input record generates one output record, either in t2 or in t3, depending on some logic in the application.
Without transactions it would be very hard to make sure that there are exactly as many records in t2 and t3 as in t1, each one of them being the result of processing one input record. As explained earlier, it would be possible for the application to crash immediately after writing a record to t3, but before committing its offset, and then that record would get re-processed (and re-produced) after the consumer restarted.
Using transactions, your application can read two records, process them, write them to the output topics, and then as a single atomic operation, “commit” this transaction that advances the consumer group by two records in t1 and makes the two new records in t2 and t3 visible.
If the transaction is successfully committed, the input records will be marked as read in the input topic and the output records will be visible in the output topics.
Every Kafka transaction has an inherent timeout, so if the application crashes after writing the two records, but before committing the transaction, then the transaction will be aborted automatically (once the timeout elapses). Since the transaction is aborted, the previously written records will never be made visible in topics 2 and 3 to consumers, and the records in topic 1 won’t be marked as read (because the offset was never committed).
So when the application restarts, it can read these messages again, re-process them, and then finally commit the transaction.
That all sounds nice, but how does it actually work? If the client actually produced two records before it crashed, then surely those records were assigned offsets, and any consumer reading topic 2 could have seen those records? Is there a special API that buffers the records somewhere and produces them exactly when the transaction is committed and forgets about them if the transaction is aborted? But then how would it work exactly? Would these records be durably stored before the transaction is committed?
The answer is reassuring.
When the client produces records that are part of a transaction, Kafka treats them exactly like the other records that are produced: it writes them to as many replicas as you have configured in your acks setting, it assigns them an offset and they are part of the log like every other record.
But there must be more to it, because otherwise the consumers would immediately see those records and we’d run into the double processing issue. If the transaction’s records are stored in the log just like any other records, something else must be going on to prevent the consumers from reading them until the transaction is committed. And what if the transaction doesn’t commit, do the records get cleaned up somehow?
Interestingly, as soon as the records are produced, the records are in fact present in the log. They are not magically added when the transaction is committed, nor magically removed when the transaction is aborted. Instead, Kafka leverages a technique similar to Multiversion Concurrency Control.
Kafka consumer clients define a fetch setting that is called the “isolation level”. If you set this isolation level to <span class="codeinline">read_uncommitted</span> your consumer application will actually see records from in-progress and aborted transactions. But if you fetch in <span class="codeinline">read_committed</span> mode, two things will happen, and these two things are the magic that makes Kafka transactions work.
First, Kafka will never let you read past the first record that is still part of an undecided transaction (i.e., a transaction that has not been aborted or committed yet). This value is called the Last Stable Offset, and it will be moved forward only when the transaction that this record was part of is committed or aborted. To a consumer application in <span class="codeinline">read_committed</span> mode, records that have been produced after this offset will all be invisible.
In my example, you will not be able to read the records from offset 2 onwards, at least not until the transaction touching them is either committed or aborted.
Second, in each partition of each topic, Kafka remembers all the transactions that were ever aborted and returns enough information for the Kafka client to skip over the records that were part of an aborted transaction, making your application think that they are not there.
Yes, when you consume a topic and you want to see only the records of committed transactions, Kafka actually sends all the records to your client, and it is the client that filters out the aborted records before it hands them out to your application.
In our example let’s say a single producer, p1, has produced the records in this diagram. It created 4 transactions.
The client, when it fetches the records from the Kafka broker, needs to be told that it needs to skip offsets 3 to 6. For this, the broker returns an extra field called <span class="codeinline">AbortedTransactions</span> in the response to a Fetch request. This field contains a list of the starting offset (and producer ID) of all the aborted transactions that intersect the fetch range. But the client needs to know not only about where the aborted transactions start, but also where they end.
In order to know where each transaction ends, Kafka inserts a control record that says “the transaction for this producer ID is now over” in the log itself. The control record at offset 2 means “the first transaction is now over”. The one at offset 7 says “the second transaction is now over” etc. When it goes through the records, the kafka client reads this control record and understands that we should stop skipping the records for this producer now.
It might look like inserting the control records in the log, rather than simply returning the last offsets in the <span class="codeinline">AbortedTransactions</span> array is unnecessarily complicated, but it’s necessary. Explaining why is outside the scope of this blogpost, but it’s due to the distributed nature of the consensus in Apache Kafka: the transaction controller chooses when the transaction aborts, but the broker that holds the data needs to choose exactly at which offset this happens.
In WarpStream, agents are stateless so all operations that require consensus are handled within the control plane. Each time a transaction is committed or aborted, the system needs to reach a consensus about the state of this transaction, and at what exact offsets it got committed or aborted. This means the vast majority of the logic for Kafka transactions had to be implemented in the control plane. The control plane receives the request to commit or abort the transaction, and modifies its internal data structures to indicate atomically that the transaction has been committed or aborted.
We modified the WarpStream control plane to track information about transactional producers. It now remembers which producer ID each transaction ID corresponds to, and makes note of the offsets at which transactions are started by each producer.
When a client wants to either commit or abort a transaction, they send an <span class="codeinline">EndTxnRequest</span> and the control plane now tracks these as well:
In the previous section, we explained that the magic lies in two things that happen when you fetch in <span class="codeinline">read_committed</span> mode.
The first one is simple: WarpStream prevents <span class="codeinline">read_committed</span> clients from reading past the Last Stable Offset. It is easy because the control plane tracks ongoing transactions. For each fetched partition, the control plane knows if there is an active transaction affecting it and, if so, it knows the first offset involved in that transaction. When returning records, it simply tells the agent to never return records after this offset.
But, in order to implement the second part exactly like Apache Kafka, whenever a transaction is either committed or aborted, the control plane would need to insert a control record into each of the topic-partitions participating in the transaction.
This means that the control plane would need to reserve an offset just for this control record, whereas usually the agent reserves a whole range of offsets, for many records that have been written in the same batch. This would mean that the size of the metadata we need to track would grow linearly with the number of aborted transactions. While this was possible, and while there were ways to mitigate this linear growth, we decided to avoid this problem entirely, and skip the aborted records directly in the agent. Now, let’s take a look at how this works in more detail.
Data in WarpStream is not stored exactly as serialized Kafka batches like it is in Apache Kafka. On each fetch request, the WarpStream Agent needs to decompress and deserialize the data (stored in WarpStream’s custom format) so that it can create actual Kafka batches that the client can decode.
Since WarpStream is already generating Kafka batches on the fly, we chose to depart from the Apache Kafka implementation and simply “skip” the records that are aborted in the Agent. This way, we don’t have to return the <span class="codeinline">AbortedTransactions</span> array, and we can avoid generating control records entirely.
Lets go back to our previous example where Kafka returns these records as part of the response to a Fetch request, alongside with the <span class="codeinline">AbortedTransactions</span> array with the three aborted transactions.
Instead, WarpStream would return a batch to the client that looks like this.
The aborted records have already been skipped by the agent and are not returned. The <span class="codeinline">AbortedTransactions</span> array is returned empty.
Note also that WarpStream does not reserve offsets for the control records on offsets 2, 7 and 9, only the actual records receive an offset, not the control records.
You might be wondering how it is possible to represent such a batch, but it’s easy: the serialization format has to support holes like this because compacted topics (another Apache Kafka feature) can create such holes.
Something we had not anticipated though, is that if you abort a lot of records, the resulting batch that the server sends back to the client could contain nothing but aborted records.
In Kafka, this will mean sending one (or several) batches with a lot of data that needs to be skipped. All clients are implemented in such a way that this is possible, and the next time the client fetches some data, it asks for offset 11 onwards, after skipping all those records.
In WarpStream, though, it’s very different. The batch ends up being completely empty.
And clients are not used to this at all. In the clients we have tested, franz-go and the Java client parse this batch correctly and understand it is an empty batch that represents the first 10 offsets of the partition, and correctly start their next fetch at offset 11.
All clients based on librdkafka, however, do not understand what this batch means. Librdkafka thinks the broker tried to return a message but couldn’t because the client had advertised a fetch size that is too small, so it retries the same fetch with a bigger buffer until it gives up and throws an error saying:
To make this work, the WarpStream Agent creates a fake control record on the fly, and places it as the very last record in the batch. We set the value of this record to mean “the transaction for producer ID 0 is now over” and since 0 is never a valid producer ID, this has no effect.
The Kafka clients, including librdkafka, will understand that this is a batch where no records need to be sent to the application, and the next fetch is going to start at offset 11.
Recently a bug was found in the Apache Kafka transactions protocol. It turns out that the existing protocol, as defined, could allow, in certain conditions, records to be inserted in the wrong transaction, or transactions to be incorrectly aborted when they should have been committed, or committed when they should have been aborted. This is true, although it happens only in very rare circumstances.
The scenario in which the bug can occur goes something like this: let’s say you have a Kafka producer starting a transaction T1 and writing a record in it, then committing the transaction. Unfortunately the network packet asking for this commit gets delayed on the network and so the client retries the commit, and that packet doesn’t get delayed, so the commit succeeds.
Now T1 has been committed, so the producer starts a new transaction T2, and writes a record in it too.
Unfortunately, at this point, the Kafka broker finally receives the packet to commit T1 but this request is also valid to commit T2, so T2 is committed, although the producer does not know about it. If it then needs to abort it, the transaction is going to be torn in half: some of it has already been committed by the lost packet coming in late, and the broker will not know, so it will abort the rest of the transaction.
The fix is a change in the Kafka protocol, which is described in KIP-890: every time a transaction is committed or aborted, the client will need to bump its “epoch” and that will make sure that the delayed packet will not be able to trigger a commit for the newer transaction created by a producer with a newer epoch.
Support for this new KIP will be released soon in Apache Kafka 4.0, and WarpStream already supports it. When you start using a Kafka client that’s compatible with the newer version of the API, this problem will never occur with WarpStream.
Of course there are a lot of other details that went into the implementation, but hopefully this blog post provides some insight into how we approached adding the transactional APIs to WarpStream. If you have a workload that requires Kafka transactions, please make sure you are running at least v611 of the agent, set a <span class="codeinline">transactional.id</span> property in your client and stream away. And if you've been waiting for WarpStream to support transactions before giving it a try, feel free to get started now.
