Diving into Exactly Once Semantics with Guozhang Wang

It has been said that in distributed messaging, there are two hard problems: 2) exactly once delivery, 1) guaranteed order of messages and 2) exactly once delivery. Apache Kafka® has offered exactly once processing since version 0.11, which allows properly configured producers and consumers to make the guarantee that each message will be processed exactly one time. 

In this episode, Kafka Streams engineer Guozhang Wang walks through the implementation of transactional messaging in Kafka in some detail, including the idempotent producer API, the transaction coordinator responsible for managing the transaction log and consumer configurations. It’s a complex topic, but he takes us through it carefully and completely.

EPISODE LINKS

• Transactions in Apache Kafka 
• Enabling Exactly Once in Kafka Streams 
• KIP-98: Exactly Once Delivery and Transactional Messaging
• KIP-129: Streams Exactly-Once Semantics

SEASON 2
Hosted by Tim Berglund, Adi Polak and Viktor Gamov
Produced and Edited by Noelle Gallagher, Peter Furia and Nurie Mohamed
Music by Coastal Kites 
Artwork by Phil Vo 

•  🎧 Subscribe to Confluent Developer wherever you listen to podcasts. 
• ▶️ Subscribe on YouTube, and hit the 🔔 to catch new episodes.
• 👍 If you enjoyed this, please leave us a rating. 
• 🎧 Confluent also has a podcast for tech leaders: "Life Is But A Stream" hosted by our friend, Joseph Morais.

Transcript

SPEAKER_02 0:00

Exactly once processing is a sometimes mysterious, sometimes controversial feature that's been a part of Kafka since the spring of 2017. We're gonna dig into that mystery and dig into a little bit of the controversy on today's episode of Streaming Audio, a podcast about Kafka, Confluent, and the cloud.

SPEAKER_01 0:34

Thanks, Tim. Thanks for having me here.

SPEAKER_02 0:36

Yeah, it's great to have you here. So Guojong is a software engineer on the Kafka Streams team. We've had a number of episodes recently dealing with Kafka Streams and a number of KSQL episodes. And sometimes we talk about KSQL. It kind of veers off into Kafka streams, but uh super important API, and it's it's when you consider Kafka not just as a big giant distributed log, but as a platform, Kafka Streams is a really important part of that whole picture of how to how to do processing over events. So what I want to talk about today uh with you, Kojong, is uh an interesting feature. It's not new, it's like a couple years old, but still much feared and little understood, and that is exactly once. So I don't know, you probably remember back when we announced this in uh what was it, May of 2017? I was brand new at the company, and I just I'd just been put in charge of the blog, and we published this blog post, and it was, hey, you know, Kafka 0.11 has exactly once, and then there's all of this discussion, extremely well-informed discussion on Hacker News, as usual, where people said, That's not possible. Exactly once isn't possible with distributed messaging. What before we even get into the feature and everything, what is it, what is that debate all about? Because I I know there are some listeners right now thinking that's impossible. Yeah. Why?

SPEAKER_01 2:02

Yeah, you're definitely right. Things like even today, like that is like one year, more than one and a half years after EOS was released in Apache Kafka OE11 release, we still see you know questions about its semantics, how it works, uh how to configure it to work in your production environment, like on a daily basis as well. Yeah, I think the first question people may ask about really the semantics of EOS is actually, really, what do you mean by processing exactly once, right? Yeah. And one common confusion that people had is that when you talk about exactly once, people may think that you actually process every record exactly once, even under you know, on any kinds of unexpected failures. And the data is actually not the definition of exactly once in stream processing, at least in Kafka streams. The definition is that the processing results are reflected to any, you know, to visible to users exactly once when you process the records coming into streams. Maybe they will be processed multiple times, but the final results will only be reflected multiple times, uh one time, exactly one time. Right.

SPEAKER_02 3:13

So that's the distinction. And we just need to get that on the table so that everybody will will have a shot at believing things you say from here on out. Um it it's so you know that there is a very robust result that in a distributed system that is passaging messages, you actually can't guarantee that messages are delivered exactly one time. I think you can correct me if I'm wrong, but I think that's just uh a corollary or a different statement of the atomic broadcast problem, which is yeah. So the if if you're familiar with the atomic broadcast problem, if you're not Wikipedia, you know, spend a few hours going down the rabbit hole, you'll be there. But uh that that is a robust result, and anybody who says we have solved the atomic broadcast problem and this is a thing that we can do is a person who is lying to you, right? That's that's actually not possible. It's like saying you have a perpetual motion machine or a square circle or something, it's just not a thing. So but people take that and that sort of loose knowledge of I heard that you can't do exactly once delivery, and they hear Kafka saying we do exactly once when we're when we're uh saying the complete thing, we'll say exactly once processing or exactly once semantics. But the shorthand is exactly once. And so everybody here, what Kujong just said. Uh what that means is as a user using uh certain APIs to move things through Kafka and to process data with Kafka, and we'll define what all that means. You effectively have a guarantee that uh messages will be processed one time. We know they're gonna be delivered more than once. This is at least once delivery, and there will be failure scenarios in which uh duplicate deliveries happen. Uh it's just that we're gonna know about them and they won't be reflected in processing results. Is that a fair restatement?

SPEAKER_00 4:59

Yeah, that's right.

SPEAKER_02 5:00

Good. Okay, so I I always anytime I talk about this, like if um I I talk about it in front of a meetup or something, which by the way, I don't do often, everybody, and I I want listeners to know this is one of those podcasts where when I ask questions of Guajhang, and you're like, oh Tim, you're just leading him, you know, this is obviously scripted. It's really not. I don't know this subject well. So this is one of those things where we are learning together, and I love this. So I don't talk about this a lot. Um but when I do, I'm always careful to make that distinction between processing and delivery. So anyway, given that, uh get us started. Like uh how how does this work? And do you want to talk about it uh from the perspective of the streams API or the the lower level producer-consumer API? Like what do you think is a better approach? Where do you want to where do you want to get started?

SPEAKER_01 5:51

Yeah, I think maybe before we really talk about how you know exactly once processing is you know implemented and achieved in Kafka streams, we can first start with talking about some common failure cases that can you know violate exactly once in stream processing generally speaking. Nice. And then we can dig from there and say how we actually, you know, how we achieve exactly once by resolving those common failures. Right. Cool. So one, I mean, let's start with stream processing in general, right? Uh with Kafka streams or with any like current off-the-shelf uh stream processing frameworks, it basically boils down to three operations. You first read some data from your upstream data, which is either stored in Kafka or in any other format. You do some processing, you transform the records, and you may actually update some of the your processing states. For example, if you are doing a count, then the whatever the stores that is storing your running count will be your processing state, right? You update the state, and then eventually you send the results to the downstream, uh, which may actually be led by some other applications or by the downstream uh system as well. So basically you have you are looking into three basic operations, which is reading the data, processing the data, and updating your states potentially, and then send the data out to your downstream. To downstream. And when you basically see a failure, the failure may happen anytime between in between those three uh in those three options, right? And when that happens, that's when you may actually have duplicates. I can give you a more concrete example. Let's say you have read the data, you have processed the data, and you have actually updated your states. But right before you're about to send it to your downstream, you actually crashed. Then upon restarting, you will actually have to reread the data and then re-update your states. Then that's actually when you will have duplicate updates, right? Because you may actually double count on some result, on some records you have been processed already, but you just failed the last uh action, which is send the results back to uh send the results to the downstream, right? Right. And in this case, as a user, you will re you will see your counting store, for example, has you know incorrect results because you have some double counts. So the the basically the summary is that because those three options, again, reading from the data and the committing that you have finished reading from this upstream, processing the red data and updating your states, and eventually sending this data back to some other streams, those three options can now be done automatic. Any failures in between those three options can cause you to have you know uh violate exactly one semantics. Got it. Got it. Does it make sense, Jim?

SPEAKER_02 8:47

It does. So we want to be able to the three steps were read input data, process records, update state is the second step. Right. And they're really synonyms for the same thing, and then send basically produce new results. Right. And we want to be able to do those atomically.

SPEAKER_01 9:04

That's right. Yeah. And there's actually a various types of failures and can cause you to not achieve exactly once. Like I gave you one kind of example. But another example I can also give you, uh, for example, in Kafka specifically, is that in your last operation, when you are about to send your results to the downstream, which is also a Kafka topic, right? Maybe the data has been successfully sent to the Kafka brokers and has been successful, successfully appended to the Kafka logs, but the acknowledgement back sent back from Kafka gets dropped on the on the floor, either because you have a network partition or some other issues. In this case, you may go try you may want to retry upon you know upon uh resuming. And the retries will also cause you to send the same results twice into the downstream Kafka topic. And hence you will also have duplicates.

SPEAKER_02 9:58

Totally, by losing an AC. So there's it's a little worse than just those three failures because there's subtlety in communicating over a network with a Kafka cluster between a stream processor and a Kafka cluster.

SPEAKER_01 10:11

Exactly. Yeah. So um basically the the mechanism in Kafka and also in Kafka streams to tackle this is to really try to try to consider how we can actually make those three operations to be really atomic. Right. And the good news is that because in Kafka streams, we actually keep a for the state itself, we actually keep a change log topic, which is uh like also a specialized topic we store in Kafka. And that can be used for failure recovery as uh as well as reprocessing, for example, as a source of truth of the state stores. We can actually convert those three operations, all of them, into some messaging uh functionalities inside Kafka. So imagine if after you updated your state store, you will append a changelog entry to the corresponding topic for this state store. Then those three operations basically become appending a changelog to the state store's change log topic, appending a record to the downstream Kafka topic, and then eventually appending a commit record to your offset topic inside Kafka. So all of those three operations become com becomes appending some records to some specific Kafka topics, and then we want to make it more atomic.

SPEAKER_02 11:31

Got it. So that now you've turned those three things into three separate produces. You're you're producing to three topics. And so now the scope of the problem is uh how do I make produces between topics atomic? Is that correct?

SPEAKER_01 11:50

Exactly.

SPEAKER_00 11:51

Okay. All right, good.

SPEAKER_01 11:52

And that that is relying on our transactional messaging uh functionality that we have also introduced in the same release of Apache Kafka OE11.

SPEAKER_02 12:01

It is almost the case that that transaction API is the same thing as exactly one semantics, right? I mean they're I know they're not literally identical, but they seem to be to me joined at the hip.

SPEAKER_01 12:12

Yes. Yeah. Yeah. And the key idea of the underlying transactional messaging is that, again, basically you want to send some messages to multiple topic and multiple topic partitions in an atomic manner, right? And what we can do is we will maintain a centralized transaction log. And this again, this transaction log will be considered as a source of the truth to determine if this batch of records which are appended to different topic partitions are considered to be uh complete and hence committed, or they can consider to be filled and hence aborted. And again, the the key idea here is that because of the single transaction log, we can basically determine that all of those messages in batches will be considered as either committed, all of them be uh c completed, or they will be aborted as a whole. So no one there will be no scenarios when some of the messages considered you know successfully cut uh appended, but some other messages considered to be aborted. Got it.

SPEAKER_02 13:17

Which is uh atomicity by definition.

SPEAKER_01 13:19

Yeah.

SPEAKER_02 13:20

Okay. So and I you uh you used a more precise word there. You said um commits uh or writes to multiple partitions. We had been saying topics before, but really it's about partitions. Um that's kind of a this is a thing Gwen Shapira said to me once is that in Kafka's mind, in the mind of a broker, topics barely even exist, right? That's a that's a naming scheme that we use. But what brokers really care about, the the actual primitive in the cluster is the partition. And so it's rights between partitions. Yeah, cool. Okay, so uh there is this transaction log, and there is um we're gonna have to dig into that, but there's one per streams application or one per stream processor, is that basically correct? Uh that's right. Okay. I'm conflating things here. You're talking about the API, and what I just asked is how the streams API does exactly once. And I know there's the transaction API. And so anyway, we take keep going. Do you want to go down the path of the transaction API a little bit more or get into streams?

SPEAKER_01 14:31

Um Yeah, I can briefly talk a little bit about the transactional messaging, and then I can talk a little bit about how we actually wrap uh all of those details in Kafka streams as a single knob.

SPEAKER_02 14:43

I feel like I want to know how to do this with the producer and the consumer. Not that I should necessarily, but uh if I understand that, I'm I think I'm more likely to understand what streams is doing.

SPEAKER_01 14:54

Yeah. So with transactional messaging, basically on the producer side, there are um a couple of transactional related uh APIs added to the producer client. As a user, you can call init transactions to basically initialize your producers to be transactional aware. Okay. And then you can call begin transactions before you send any data within this transaction.

SPEAKER_02 15:19

Okay.

SPEAKER_01 15:20

Just think about if you are familiar with RDBMS uh you know syntax, you have begin transaction, and then you can do reads and write, blah, blah, blah, and then you will call commit transaction.

SPEAKER_02 15:29

Yep, feels like home.

SPEAKER_01 15:31

Yeah. We designed the APIs actually just to be similar to the RDBM semantic as well, where you can call begin transactions, and then you can call produce, and you can actually produce a lot of messages to different topics and to different topic partitions from the broker point of view. And by the end of it, you can call commit transactions to basically commit all the messages you have been produced in the previous API calls. Or you can also call abort transactions. For example, if you capture some exceptions during this process and you want to basically abort the whole transaction so that none of the messages will be visible uh to a read-committed consumer down the stack.

SPEAKER_02 16:13

Yeah, good qualification to a read-committed consumer. Uh we'll figure out what that means in a minute. So that at the produce side, I have to tell my producer uh get ready for transactions, you know, be be transactional, and then begin and then produce, produce, produce, and then commit. Yes. So and I know we're gonna have to go back and forth between producer and consumer here, but what is the producer doing differently when I do that?

SPEAKER_01 16:38

Right. So first of all, you know, when you call produce, right, all the messages are indeed physically being appended to the brokers, right? The uh you know, basically this appending message actions cannot really be revert when you really want to abort a transaction. What we actually make it work is to let those messages to be not visible to the downstream consumers if the consumer is configured with read committed. And I know that I'm maybe slightly switching from producers to consumers, but I think this is also actually a very important concept on the you know on the reading side, which is that again, physically, you actually will have both aborted messages as well as committed messages in the KAFA brokers. And if the consumer is not configured as read committed, all of those messages will still be uh sent back to the consumers and be visible to the user.

SPEAKER_02 17:32

Okay. That makes all the sense in the world. They have been produced.

SPEAKER_01 17:35

Yeah. If you configure the consumer to be read committed, then actually the consumer client will be able to filter those aborted messages when read from the brokers and not actually return them from the consumer API call. So that from a user perspective, they will only be reading messages that are committed only. That makes sense. Yeah, this is actually also like a very common confusion, I would say, in practice, because a lot of users they ask in the AK meaningless saying, hey, I have already enabled it exactly once. Why I'm still seeing uncommitted data or aborted data in my final stage when I consume it. And uh you know, I would say 90% of the time they actually forget to configure their consumers to be re-committed. There you go. There you go. So this is actually very important to make it c uh clear. Right, right.

SPEAKER_02 18:29

It is not as if this uh this new API causes committed produced messages to be deleted from topics, because that's not a thing that a broker is ever willing to do. Yeah, yeah, exactly. This is this is actually just doing more writes to another topic, which I think you're gonna get into next. So on the produce side, you talked about a changelog topic. Is that happening? Is the the transactional producer doing that?

SPEAKER_01 18:56

Yeah, the changelog topic, again, it's basically uh only used for Kafka streams to basically maintain the you know, the change capture stream of the update history of the state specific state stores. So when you are updating the state store, the producer will also write write to the corresponding changelog within the same transactions as it writes to other Kafka topics. And with within the same transaction, the producer is also uh able to commit the read the red offsets uh by reading uh by writing it or appending it to be more specific to this uh offset topic at the same time. Okay. So all of those will be done within the transaction.

SPEAKER_02 19:39

And that's but you said that's just in the Kafka streams case, uh the change that topic. Okay, so then uh let's keep walking through. We're just uh let's we'll come back to that. We're still just in producer land here. And I've initted transaction, I've begun transaction, and I've produced to six of my favorite partitions. Um what is different about that producer though, and what what is happening? Uh so forget I said anything about the changelog topic. What is it doing?

SPEAKER_01 20:08

So the producer will actually talk to uh the broker sign module. We have a additional module added in uh O11 called the transactional coordinator. Basically, you can imagine that the producer will register to the transaction coordinator saying, hey, in this transaction, I'm um writing to those following topic partitions. And the transaction coordinator knows who are the leaders of those topic partitions. So he will actually also remember such metadata on the broker side as well. So you can keep uh producing to more topic partitions, and the producer client will automatically communicate this information to the transactional coordinator, basically saying, hey, I'm adding more and more top partitions into this transaction that I'm producing right now.

SPEAKER_02 20:54

Okay. And then as as those new partitions pop up in the transaction, the producer tells the transaction coordinator Yeah. Uh okay, so you were going to say keep keep going. I got more questions.

SPEAKER_01 21:06

Yeah, and then all of those is uh abstract away from the producer users, of course. Of course. This is happening behind the scene. And then when you commit your transaction, the producer will basically send a request saying, I'm able to commit this transaction right now. And the transaction coordinator is then responsible for telling all the leaders of the partitions that is within the transactions by writing a so-called transactional marker, saying that, okay, for this transaction, all the messages are now being treated as committed, and then they can be, you know, they can be fetched to the downstream consumers as well. Got it.

SPEAKER_02 21:45

So and that transactional marker is a message produced effectively to each of those partitions that is involved in the transaction?

SPEAKER_01 21:54

Exactly. Okay. It is a specialized internal message so that only the brokers know how to interpret it. It will never be returned to the clients. Got it. It's a way of interbroker communication. Of course.

SPEAKER_02 22:06

That makes sense. Is there an equivalent begin marker?

SPEAKER_01 22:09

No. There's only a transaction marker, but the marker has a field which is indicating either it is a committed transaction or it is a com uh aborted transaction. So basically right after this marker, any new data can be considered as the starting of the new transaction. So that's why we don't need a begin transaction marker. We only need a you know and a transaction marker.

SPEAKER_02 22:30

Okay. Okay. So you are um there's a begin method in the API, but you're implicitly in a new transaction after a previous one is committed. Right. Got it. Okay. Okay. So and that transaction controller that makes one think of a two-phase commit sort of thing. That that there's this other node. And that's a that's a broker? That's a like basically a a function that a broker performs? That's right. Okay. That's right. Is it Oh, go ahead. Yeah, go ahead. Always the same broker? Is it a thing that gets elected, you know, just depending on the lead partition you start with?

SPEAKER_01 23:09

I mean, how do how do we know Yeah, it's elected. Uh basically it uh piggybacks on the leader partition of this specialized transactional log, which is also an internal uh topic, just like the offset topic, right? So basically all the brokers will be the transaction coordinator for some transaction groups. And not not like we it only like a single broker to basically handle all the transactions. Got it.

SPEAKER_02 23:33

Okay, so the lead of the partition that you're writing to uh or the your partition of the transaction log uh topic, uh whoever is leader of that is your transaction controller. Yeah. Okay. So you put it in the room.

SPEAKER_01 23:51

And again, the transaction log topic here is used as like really the centralized place for maintaining the states of all the ongoing transactions and also acts as a source of the truth. Right. Like imagine if you have a broker side failover, like the current existing transaction coordinator for some of the transactions has been crashed. And another broker starts to take over the responsibilities. How he would know who are the current ongoing transactions, who are the committee transactions, right? They will also read it from this transactional log. And after that, basically, this broker is up to date on all the ongoing transaction status. He can continue finishing the uh the work, just like you know, sending those transactional markers, for example, to the partitions that are within a committee transaction. Okay.

SPEAKER_02 24:42

And is that and so uh again, uh this is not the same as the uh transaction log that Kafka Streams maintains that we're gonna talk about later. But there is this topic that and what what's the proper name for it again? The K.

SPEAKER_01 24:58

It's called the transaction log.

SPEAKER_02 25:00

The transaction log. And is that a topic of state for the various extant transactions in a cluster, like a like a compacted topic, or is it properly a log?

SPEAKER_01 25:12

It is a log. It's uh just like an offset topic, right? So currently in Kafka, we have two internal topics which are not like exposed to the users. Right. One is called the offset topic. Although the name is called offset topic, it actually is used to store both the offset information as well as the consumer group information as well. Right. And it's uh it's a basically a specialized compacted topic. And we have this transactional topic which is added in O11. This is also a compacted topic uh as well.

SPEAKER_02 25:42

Okay, okay. It is the we could think of it at this point as the current state of of transactions that uh any transactional producers have running. That's right. Got it, got it. And that again, so that's distributed state that that distributing that transaction information is a distributed state problem. But as per the usual strategy with Kafka, when we need to manage distributed state, we use Kafka to manage that distributed state. Because by definition, we have a Kafka lying around and Kafka's real good at that. So that makes sense.

SPEAKER_01 26:17

Okay. Yeah. And actually, just to clarify one one more thing, uh Tim, is that although you know the implementation may sound like you know two-face committing or two-face locking uh in the database world, actually it is not really the same idea of uh 2PC. Because again, the key point here is that we have a centralized place, a single log, which is the disk transaction log, used as a source of truth for all the current ongoing transactions. So actually it's its key idea is more similar to what we call it, that the non-deterministic or sequential transaction commits protocol, uh, as we talk now, compared with 2PC. Because it actually by using this transaction log, we exclude any you know non-determinism as well as you know out of ordering transactions, because all of this is actually executed following the order of this transaction log as well. Got it, got it.

SPEAKER_02 27:14

Yeah. Um I was I was expecting you to react negatively to me saying two-phase commit more quickly than you did. So I admire your your self-control. Uh I I don't want to say I was trolling you. I wasn't trolling you, but I knew it wasn't that. And um I knew enough to know that, and I'm glad that you clarified. Because I I I was in the process of getting lost in the discussion and not going to come back and clarify that, hey folks, this is a two PC. It actually performs. Um we'll talk a little bit about performance. I I really want to go through the semantics and not not benchmarking in this discussion, but we can talk some about performance when we're done. So stick with us, folks. All right. So that's the produce side. Tell me about the idempotent consumer and how is that related to all of this?

SPEAKER_00 28:00

Oh, yeah. Uh on the consumer side, we're producer. I said consumer, idempotent producer. Producer. Yeah. Okay.

SPEAKER_01 28:06

Yeah, the idempotent producer is uh you can consider about that as you know, semi-octogonal to the transactional producer. The idempotent producer only tries to resolve one specific problem, which is the retry, causing duplicates problem. And this is one of the most common issues that can violate the, you know, uh violates exactly once uh delivering in Kafka as well. Got it. So the retry issue that I just mentioned at the beginning is that when you when you send a record to Kafka, right? The record has been successfully handled and it's successfully been appended to the Kafka logs physically, right? So it cannot be reverted anymore. But it's just that because either Kafka brokers fail to send the response back, or he sends the response, but the acknowledgement response is not being successfully delivered back to the producer. At this time, the Kafka producer typically will retry sending the data again, which will cause a duplicate. And the idempotent producer is designed to specifically fix this issue so that we will not actually cause duplicates even if we retry. Got it.

SPEAKER_02 29:13

And how does that, I don't know if we if it if it's too much detail, stop me and we'll back off. But uh maybe you can explain to us how that works. Like what when when uh what does idempotent producer mean? Because I can imagine doing something stupid like a topic where I have a key that's an account number and uh a message that is a command to increment something. Uh that's terrible. Uh but it's not particularly Adam Potent. So what does this mean?

SPEAKER_01 29:44

So Adempotent producer, um yeah, without drilling into too much details and that you know terminologies here, you can think of it as that uh when the Adam potent producer is sent to a specific broker for pending data, right? That broker will actually remember this producer. It will not actually remember any uh you know, non-etempotent producer, it will, but it will remember its producer and remember its current produced sequence, I would say. Okay. Let's assume that the producer has sent like 10 messages. Those 10 messages will be actually numbered and be remembered on the broker side. So broker will know, okay, this producer named, you know, PA producer IDA, let's say, has produced up to sequence number 10 at this time. Okay. And it has been successfully appended. So I know that the next sequence number I should be expecting from the producer will be 11. Then when producer does not receive the acknowledgement of, let's say, the last produced message, and then when it retries, it will say, okay, I want to retry for sequence number 10. But at this time, because the broker actually remembers this producer along with his ID and the current sequence number, it will actually reject this produced request by saying, Hey, this sequence number 10 is already successful appended on my log, just so you know. And the producer, upon receiving this error code, knows that it does not need to retry anymore and can then proceed to send the next message. Got it. That's why the, you know, by we actually achieve a deponency on the product on the broker side.

SPEAKER_02 31:17

So that implies that producers have uh a unique identity, and each one on a per partition basis, I'm guessing, maintains those sequence numbers just as state on the producer side. Yes. Okay. Uh and is there any other state that you have to manage in the producer now? Sounds like there is. You have to remember a little bit of what you have sent.

SPEAKER_01 31:39

Yes. You have to remember, like for an important producer, you basically have to remember, like I said, the the net the sequence number of the message you have you sent so far, as well as the sequence number of your in-flight requ uh produce request as well. Like if you have already been acknowledged on offset, sorry, on sequence number 10, and you are sending sequence number 11 to 15, but have not been acknowledged back, you have to all remember those numbers on the producer side. Aaron Powell Got it.

SPEAKER_02 32:09

But that's usually small. I mean, that's uh going to be governed by the max in flight requests configuration parameter, which is usually small. So I just wanted to go through that. If anybody sounded if that if anybody's getting scared about this massive amount of state we were suddenly accumulating in the producer, uh it's not like you need a day's worth of messages. I mean nobody does that with flight requests. That's not how Kafka works. So it's it yes, it's state. No, it's not really a daunting problem. Cool. Okay. So I I feel like I kind of understand what's going on in the producer. Shall we move over to a read-committed consumer?

SPEAKER_01 32:44

Yeah, of course. So actually on the consumer side, um it's it's much more simpler than the producer side. Um like I said, so uh on the consumer side, we have added an additional config called the read mode. And by default, it is uh uh read uncommitted, which means that it will read all the messages, no matter if it is uh you know a non-transactional message or it is a transactional message but it is a boarding message. Uh basically all the messages will be returned from this consumer. Uh but the users can uh choose to override this config to read committed mode. And by reading committed mode, the consumer will, if the if the return messages from broker side is a transactional message, the consumer will determine if this message uh can be returned back to the user. And uh the brokers will also do the filtering on its end when it receives the fetch request. And the fetch request is marked as okay, this is actually coming from a consumer that is in read-committed mode, so that it will not return transactional messages that are not yet determined whether it is committed or aborted back to the customer.

SPEAKER_02 33:55

Uh-huh. Because we have the end marker. Okay. So the read request indicates whether it's read committed or not. And if it is, then and the broker sees messages which are are committed in the log sense, but there's no transaction marker, then they don't come back.

SPEAKER_01 34:13

Yeah. And in order to actually preserve the you know the offset ordering, right, the brokers can only return up to the messages where all of its previous message has either not transactional or it is transactional and its state has been determined. Right. What I meant is that let's say if you have uh messages of offset 0 to 10, and uh from 0 to 5, all the messages has been committed. But offset 678 has not been determined whether it's committed or not. And offset number nine is determined committed. In this case, the broker can still not return the message of offset number nine because uh you know it cannot determine whether the messages in between, namely 678 can be returned or not. And then in order to maintain the offset ordering, uh it will actually only return up to offset five. Got it.

SPEAKER_02 35:09

That of course that makes sense. Now that's that's just a single partition, though. At least I understand that in terms of a single partition. So bring us out a little bit.

SPEAKER_01 35:16

Uh sorry, can you say again?

SPEAKER_02 35:18

Yeah, yeah. Um I that makes sense for a single partition, but um I and I whoa whoa, I suppose for a consumer now, maybe that that was a that was a silly question. Uh is that it for the consumer? Is there more to the story on the consume side than what you just described? Or is that that's it? For uh for multiple partitions? Yeah, but I guess if I'm a consumer, there's no such thing.

SPEAKER_01 35:41

Yeah. Yeah. Actually, again, the broker will decide what messages to return based on you know the commit mode and based on the current, you know, uh transactional markers on its own partitional uh partitions. Yeah. So and then all the brokers will make this decision independently. So there's no coordination.

SPEAKER_02 36:00

None because that coordination has already happened because the transactional marker didn't get committed until the transaction controller on the produce side was satisfied that the transaction could be committed. There you go. There you go. Um Yeah, you're right. That is a lot easier uh to read than uh than to write. Yeah. Okay. Now, if that makes sense, could you bring us up to Kafka Streams? Because this this or this uh notionally is a podcast about exactly what's processing in Kafka streams. But it like you say, we really have to get this groundwork in place before that makes any kind of sense. So how does at at the at the produce and consume, I just want to summarize if you're using the producer and consumer API and you want exactly one semantics, there are some method calls you make on the produce side. You named three of them. They were init, begin and commit, basically. And on the consume side, you uh set you configure yourself to be read committed. Um and that's pretty much it. So when I am using Kafka streams, and we can just assume uh everybody basically knows what that means, this functional high-level API for doing stream processing in Kafka, which itself is under the covers using the producer and consumer API. How does all this interact?

SPEAKER_01 37:23

So in Kafka Streams, it's um even simpler. We only expose a uh single config, uh, which calls the processing guarantees. And again, by default, it's the its default value is at least once. But when user chooses to override this value to exactly once, Kafka Streams will set uh you know correspondingly on the embedded producers to be transactional producers and use its APIs like init, begin, commit, and above transactions, of course, to send messages in a transactional manner, and also will configure its embedded consumers to be read committed so that it will only read committed messages from the previous uh upstream uh topic as well.

SPEAKER_02 38:08

There you go. So basically it just uses, since it's it's producing and consuming for you, just uses those APIs for you.

SPEAKER_01 38:16

Right. Right. Of course, there are a lot of like devils in the details on the implementations, but that's basically the high-level idea of how streams leverage the transaction messaging to do that.

unknown 38:26

Right. Yeah.

SPEAKER_02 38:27

That makes sense.

SPEAKER_01 38:29

Yeah. Um maybe one more thing to add is like I mentioned, you know, uh we actually effectively transform those three operations. Again, the reading from the message uh from the incoming uh stream, updating the state as you process the record and send to the downstream all of them into sending to some sort to some topics, right? We have to basically make sure that they are sending through the same producer client for each of those stream tasks. Okay. And the producer will actually use the transactions effectively to say, okay, whenever I commit the state of my processing progress, I will make it align with my transactions as well, with uh when I send messages. Uh those are the key ideas how streams will implement that.

SPEAKER_02 39:18

Got it. So that first step committing the state of my uh stream processor, that's any operation that would be stateful, like if I am grouping an incoming case stream into a table and you know, performing some or you know, group stream and then performing some aggregation and getting a table or doing a stream to stream join or any of these things that requires state in the stream processor. Some things that Kafka Streams does are stateful. In fact, most of the interesting things that Kafka Streams does are are stateful. Uh that's one of the massive pieces of value you get out of the API. Uh and you don't ever owe a streams programmer. You don't want to be in the business of managing distributed state. That's an infrastructure problem and it's a waste of your time if you're an application developer. So Streams manages that state for you. And one of the ways it manages that state is basically if you just think of it maintaining this key value, kind of big hash hash table in memory, it will persist that back to a compacted topic in the Kafka cluster so that if your stream processing node goes down, you can rehydrate from that topic. So that's what Wu Zhong was talking about there. And you're saying at the time that you commit that state, because that's gonna be a periodic thing where you say, here's my uh big uh key value store that I'm using to keep track of the stream stream join. And every so often I'm gonna make sure that the cluster is up to date with my state store. When you do that, you also make sure transactions are what? That take me through that again.

SPEAKER_00 40:57

Yeah.

SPEAKER_01 40:57

So when you do that, uh again, you will I mean, be actually before you do that, you'll have already started your transaction so that all the messages you send, either to the downstream topics or to the changelog topics when you update your states, or to the offset topics when you commit on your incoming uh fetched uh records. They will be all within the same transactions already. So when you say, okay, I'm ready to commit my current processing state, it will basically translate that into a commit on the underlying transaction messages as well.

SPEAKER_02 41:32

Okay. And it gets to decide that because uh well, because it's it's the one that's managing the semantics of the stream processor anyway. Uh you as an application developer are saying things like, you know, you want to filter using this lambda and aggregate using this lambda and you know, join this to that and make a table out of the other thing. And you know, those are the primitives you're using in the streams API, but you're not at the streams API level, there simply are no meaningful transactional primitives that you interact with. The API can uh dictate when those commits happen because you yourself are not doing any of those. Yeah. Being a functional API, there would there would be no meaningful place to do that. That since those transactional begin commit abort things are fundamentally imperative.

SPEAKER_01 42:23

Aaron Powell And also if uh if there's any like exceptions or errors happening during the processing. For example, if it's either a you know, a code bug, like you divide it by zero, or you have some uh you see some like error exceptions when you try to write to your you know uh say local state stores, or even like if there is say a rebalance happening and uh you you're do not own you do not own the tax anymore, and you realize that as a task has already been migrated, all of them will also be handled automatically by streams by aborting the current transactions so that none of those uh you know changes uh will be reflected when you resume processing them. Got it, got it.

SPEAKER_02 43:04

And then the stream stream processor will automatically uh since those transactions aborted, it'll roll back to the beginning of the transaction or the the messages after the last successfully committed transaction, and you'll reprocess those messages. So the listener take note, yes, you received the messages twice. The node got them more than once because there was a failure, and so there's duplicate transmission of the message, but the result will never be propagated. If you remember Gujong's three-part thing, you read, you process update state, and then you send results downstream. Well, you never sent results downstream. So it's okay to re receive and reprocess and re update state because the act of rolling back means you go back to what the state was at the beginning of the last committed transaction, and you go back to the first message after the last committed. Transaction and yep, you get it twice, and you get to do it all twice. And you hope that doesn't happen a lot because it would get expensive, but in terms of the correctness of the system, it is in fact exactly once processing.

SPEAKER_01 44:13

Yep.

SPEAKER_02 44:15

So final final question. What and I don't know if you have an answer to this. What has been in your mind the most interesting part from an engineering and you know from a computer science perspective about this whole effort?

SPEAKER_01 44:30

Well, I think the most interesting part is really like when we start the basically the design principles of achieving exactly once here. Because, like you said, there are already multiple solutions in the literature, either from the academia or from the industry that how to achieve, say, basically data consistency in a distributed system, right? And in Kafka streams or in Kafka generally speaking, world, we actually leverage a lot on the logs of Kafka by itself and leverage on the ordering guarantees it maintains to achieve to tackle this problem. And like I mentioned before, the key idea here is that because we are relying on a specialized transactional log, which is ordered and it's highly replicated and hence highly available, we can actually rely on that to basically determine the current states of any of the ongoing transactions, and we can just execute them following the orders of the transaction log. So I think that is actually the most interesting part throughout this exercise and how we actually enable this key idea into all the implementations in around Kafka. And uh the whole project, I mean, takes a very long time. I think before the O11 release, we took at least three engineers to spend uh more than three quarters to finish from the design to the implementation. But the key idea behind this implementation never changes. And uh I think to me it proves to be really like a nice and elegant, uh elegant design principle. And uh, you know, again, if anyone is interested in learning more about that, uh we have a couple of series of uh blog posts as well as the wiki docs around this uh implementation on Confluent and as well as Apache Kafka community.

SPEAKER_02 46:23

And those are uh very good reading. Uh some of them are quite in-depth. Uh the uh the blog post from a year and a half ago, that was a beast of a blog post, but uh rewarding and it really takes you through if you want to go through all this in a different way, in written form, with a few diagrams, highly worth it. Uh, I'll make sure that gets into the show notes. And um, yeah, that's that's reading I recommend. Our guest today has been Guozong Wang. Guizhang, thanks for being a part of Streaming Audio.

SPEAKER_01 46:52

Thank you. Thanks, Tim. Hope I can come here next time.

SPEAKER_02 46:55

And there you have it. I hope that was helpful to you. If you've got questions, you can ask me at TL Burgland on Twitter. That's T-L-B-E-R-G-L-U-N-D. Or you can leave a comment on any of our YouTube videos. Your question might be featured on the next episode of Streaming Audio. And feel free to subscribe to our YouTube channel and this podcast wherever fine podcasts are sold. And if you subscribe through iTunes, be sure to leave us a review there. That helps other people discover the podcast and just generally helps us get the word out. We appreciate your support. See you next time.