Confluent Developer ft. Tim Berglund, Adi Polak & Viktor Gamov

Fault Tolerance and High Availability in Kafka Streams and ksqlDB ft. Matthias J. Sax

• Confluent, original creators of Apache Kafka® • Season 1 • Episode 109

Share episode

Use Left/Right to seek, Home/End to jump to start or end. Hold shift to jump forward or backward.

0:00 | 54:03

Apache Kafka® Committer and PMC member Matthias J. Sax explains fault tolerance, high-availability stream processing, and how it’s done in Kafka Streams. He discusses the differences between changelogging vs. checkpointing and the complexities checkpointing introduces. From there, Matthias explains what hot standbys are and how they are used in Kafka Streams, why Kafka Streams doesn’t do watermarking, and finally, why Kafka Streams is a library and not infrastructure. 

EPISODE LINKS

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.
SPEAKER_01

Stream processing is hard. Default-tolerant, highly available stream processing is even more so, and building a system like Kafka Streams presents a bewildering array of trade-offs in solving the problems that present themselves. I talked to Matthias Sachs, a Kafka committer and active contributor to Kafka Streams, about issues just like these on today's episode of Streaming Audio, a podcast about Kafka, Confluent, and the cloud.

SPEAKER_00

Thanks, Tim, for having me. Glad to be on the show.

SPEAKER_01

Great to have you here. Now you are, uh correct me if I'm wrong, an engineer who works on uh Kafka Streams. Is that uh more or less correct?

SPEAKER_00

Yeah. You could say that.

SPEAKER_01

Tell us a little bit about what you do. And I I always like to ask uh how you came into this role. Like what work did you do before that that prepared you for this?

SPEAKER_00

Yeah, so before Confluent, I was actually at the university um doing some research on data stream processing in general and distributed systems. Back in the days, I basically did not do much with Kafka. I was was getting started with Apache Storm because that was the de facto standard back in the days. Um and so at the university, there was also this research project called Stratosphere that eventually turned into Apache Flink. So I was also involved in Apache Flink. Um and well, then I got the job at Confluent and started to work on Kafka and Kafka Streams, which was of course pretty exciting because building something new from ground up, even if I was not there from the very beginning on, was of course very exciting. Um and then later I also got a little bit involved in KSQLDB. What is kind of natural as Kafka Streams is the runtime of KSQL. Um and then, of course, well, when you work on Kafka, you also get involved in two more broader things, working a little bit on clients, understanding the brokers better, and you know, learning about the whole ecosystem.

SPEAKER_01

Everything uh works out to be sort of connected after a little bit. Yeah, absolutely.

SPEAKER_00

Absolutely. Uh I mean you need to understand a couple of basics, for example, how does rebalancing work in detail in order to leverage it to build Kafka streams in a reasonable way.

SPEAKER_01

Yeah, okay, that makes sense. You'd have to know quite a bit about rebalancing uh in order to do things like failover and scale out and uh and things like that. That maybe we can maybe we can get into that a little bit in the in the discussion.

SPEAKER_00

Yeah, correctly. Another good example would be would be exactly one's processing semantics. So you really need to understand how Kafka transactions work in order to leverage the producer and consumer APIs correctly in Kafka streams to have this kind of you know end-to-end processing semantics.

SPEAKER_01

Yeah, that's that's uh also a good point. That's a configuration setting in your streams application. You know, you you set uh exactly once on, and everything just works. You don't have to worry about it so much uh like you would if you were doing it with producer and consumer. But somebody had to do that worrying, and that somebody in part is you in this case.

SPEAKER_00

Yeah, I was I was involved in implementing that, yeah. That's correct. Cool.

SPEAKER_01

I want to talk about fault tolerance and availability. I mean, you mentioned rebalancing, and I briefly mentioned failover and and scale out. Um and these are features that uh when we talk about Kafka streams, they're they're they're features that we extol in the the API. Uh you know, it it famously solves uh some state management problems for you that the consumer group rebalance protocol and just the whole consumer group mechanism doesn't really solve for you. So these really nice things that are built into Kafka streams, and uh we'll link in the show notes to a previous episode that that talks about those basics if you're not up to speed on Kafka streams. But I'd kind of like to deep dive into some of these things uh that have to do with the just broadly speaking, what we call fault tolerance and availability in Kafka streams. So to start with, help us out with the difference between those things. If you could define fault tolerance and define high availability and and how they're different.

SPEAKER_00

Yeah, yeah, sure. Um I mean fault tolerance is basically a requirement to be highly available. And and fault tolerance in the very generic definition just means if something goes wrong and there is an error, then the system can recover from it. But it does not mean that the system is not offline while it is recovering. And this being offline or being online all the time, then it's a kind of difference to high availability, because high availability says, well, even if there is an error, um, the system is still able to operate and do something and basically recover on the site. And then, of course, there is this kind of notion of you know how high available a system is, and then it's kind of well, there are you know things like mean time to recovery, and there are different categories, how you can categorize this. For example, a system could be still available for reads, but could not be available for writes during recovery, or it's fully available for reads and writes, and then you can you know make different categories and say, you know, what kind of high availability the system um offers. And of course, usually if some error happens, there will always be a tiny bit of offline time for some features because well the failure must be detected or whatever. Then it also depends on this kind of you know how long is this period? Is it a couple of milliseconds, is it a couple of seconds, is it five minutes? Uh, and then you also at some point draws a line where you say, well, if it's five minutes, maybe it's not highly available anymore. If it's like 10 seconds, yeah, that's still something we can call highly available, depending on the application's needs, obviously.

SPEAKER_01

Yeah, they they really are uh not precise terms. They're almost marketing slogans. Uh, I mean, we use them uh as engineers to talk about systems, but uh you have to give them precise definition if you really want to be uh if if you really if you have, I guess like tight tolerances you have to engineer in a system because high availability, as you say, like something might need to time out and fail over uh and it takes 20 seconds or something like that, and and then it fails over and it's fine. Or you know, maybe there is some kind of rebalance that has to happen within a cluster and data needs to get moved around, and you never lose data. And like maybe you can still take writes, but reads are inconsistent during that failure, just making this up. You know, these can be failure mode characteristics of a given system that describes itself as highly available or fault tolerant. And you have to dig into the details of really what you mean by those things. Uh it describes a sensibility and like a set of goals of the system, but its actual failure characteristics are uh those are like more specific facts that that aren't slogans, but they're like, you know, when this thing happens, it it can take between this long and this long to recover.

SPEAKER_00

Yeah, absolutely. So always always go with the fine print. Usually it's it's documented what it actually means. And then usually you also have some some options to configure a system to you know choose between different trade-offs. Um and that is of course highly, highly application dependent. That is the reason that most systems provide you with some trade-offs, because the system cannot pick it for you because they don't know the system doesn't know the requirements of your specific application.

SPEAKER_01

Right. So true. Um and I know, like in the case of distributed databases, uh, we talk about uh cap theorem characteristics, and we say, well, you know, they're always partition tolerant uh because they're distributed and partitions are going to happen, and so you have to choose between consistency and availability. But when you actually read the fine print, it is rarely a binary choice between consistency and availability. There are these configurable and uh finely shaded gradations between C and P or rather C and A that are meaningful and are like valid engineering choices that you have to make as a designer of a system.

SPEAKER_00

Yeah, totally agree. It's it's it's really uh wide spectrum.

SPEAKER_01

Yeah. Um now I I mean I want to get into how Kafka Streams and KSQLDB handle these things, but there are a few, I think, sort of definitions we have to get through first, like uh change logging. Tell us about changelogging and its role in satisfying these two properties of a system.

SPEAKER_00

Yeah, so change logging, as the name indicates, basically means that we're going to log all changes. Um and this is basically related to changes to state. So Kafka Streams is uh is a stateful stream processing library and has built-in state management. And so in order to maintain the state, Kafka Streams is using the change logging technique. It basically says every time the change is updated, we also write it into a log to record the change. Um, and of course, because we're using Kafka, this log is going to be a Kafka topic. Um and that helps to basically um give fault tolerance. Um, change logging is a very old technique, basically, it's also done internally in all relational database systems, also non-relational database systems. There, basically, you have this what's very often called uh a write-a-head log or a commit log, where the database system basically records all changes that it's going to apply in a transaction. And if the transaction commits, um then it could actually replace a log to update the state in a consistent manner, so the tables in this case. And so Kafka streams is doing the same thing. Um, when state changes are applied, then we write all those changes into this changelog topic into the Kafka cluster. And the Kafka cluster, of course, is highly available, and it also can replicate the changelog topic for us, so we're never gonna lose it. And then we also apply the changes to the local state in your replication. But the the state in your application is kind of ephemeral, obviously, because well, the client could go down. And if that happens, then we can go back to the changelab topic into the Kafka cluster and reread the changelab topic to actually recover the state on the client side for you.

SPEAKER_01

Excellent. So um it really is I appreciate you pointing out that uh this isn't a Kafka or a Kafka streams thing. This is a everything that stores data thing. Um you you would be hard pressed to find a database that did not do commit logging or write-ahead logging, um, which is to me, uh and I I don't I don't think I ride this hobby horse much on the podcast. I I know I I talk about it a lot, but not too much on the podcast. But it's one of my favorite architectural insights in the systems that have emerged around Kafka, you know, where people have really uh embraced event streaming in a deep way. Uh you get that you know, commit log at the center of your system, sort of commit log writ large, uh, where your services are doing change logging and communicating through change logs, more or less event streams change logs, two sides of the same coin. Um but just that insight that that is a that is a feature of all databases and increasingly really a feature of all distributed systems uh that need to maintain state. The best way to do that is just remember the changes to state and then um create views of snapshots of state and and reason about it later on.

SPEAKER_00

Yeah, yeah, absolutely. I mean mean change logging is of course just one way how you can do that. Um but but change logging has a has a lot of very nice properties um because in the end, I mean recording an individual change is quite cheap.

SPEAKER_01

Right. And you uh I think another important thing about it is you don't need to make many decisions about what the change means. Uh if if you're always materializing a current view of state, you know, you've got the event, you're always processing the event, um, but you kind of have to prematurely optimize your interpretation of that event and then materialize the state somewhere, like in a table. Um and you don't you don't always make the right decisions up front about what that event means. But if you're just primitively storing the raw event, then later on you have the opportunity to reconsider what it means since you've got it there, uh, or make multiple interpretations of the event uh in multiple views. So it's it's you know, um probably if you're listening to this podcast, you're already sold on that idea. But if you're not, you should be. It's just a good thing. We're kind of also dancing around this idea of checkpointing. Uh you like you very nearly started to talk about that. So uh you define changelogging for us. Tell us tell us what checkpointing is.

SPEAKER_00

Yeah, yeah. Checkpointing is is a different mechanism to uh provide for tolerance. So the idea here is to say, well, um, instead of maintaining a change log, um at certain points in time, uh I basically copy the whole state over in a consistent manner uh and store it somewhere else. And if something goes wrong, then I can just reload the whole the whole checkpoint and continue processing. Um and and checkpointing, especially in a distributed system, is rather complicated because well, if you if you have your stream processing program that is expressed as a data flow graph, then well, you know, data is flowing through all those stages in parallel. And then it's kind of well, what does consistency even mean here? And how can you get a consistent snapshot? Because that's important. So conceptually you can think about this. You would say, well, at my source nodes, I stop to fetch new data in my data flow program. And then I wait until all in-flight data is fully processed. So you could think about like draining your data flow program. And if that is done, then basically no state changes are applied, and you know that all the individual pieces, because the state is of course sharded and spread out across the application, is kind of in a consistent state with regard to your input. And now you could say, well, now I copy the state somewhere else, um, for example, into uh an S3 bucket, or you have kind of an HDFS system running somewhere. Um, and what I also do is I also record in the term of Kafka the offsets of my input topics. And then I have basically this consistent state where I know, well, I have processed up to offset XYZ in my input topics, and the current state with regard to this offset is also stored now in S3, for example. Um and this gives you this level of consistency you need. And then you can continue processing if you wish. And now, if something goes wrong, what you would do is you would say, well, I tear down the whole program, I restore the checkpoint, and I also rewind my sources to the exact offsets I recorded. And then when I when I resume processing, of course, because the data in the checkpoints corresponds to the input data, I basically reprocess everything before the failure happened. Uh, and then processing continues in uh in a consistent manner. Of course, now the problem is that you don't want to really stop processing to actually do a checkpoint. And then here's the magic basically comes in.

SPEAKER_01

So what uh tell us about how that actually works in Kafka streams. How does how does this relate to the way Kafka streams handles?

SPEAKER_00

Yeah, in Kafka streams we we don't really have checkpoints, so so we don't need this. Um so and so there's this this basic difference. So if you if you do this checkpointing as I just described, you basically would get like a global consistent checkpoint. Um and in Kafka streams we don't do that, um but to provide for tolerance and high availability, you also don't need that. Kafka streams programs themselves are basically split up into multiple sub data flows that we call subtopologies, uh and those are executed in parallel in tasks. And so basically, Kafka streams provides task tolerance, if you wish. What that means is, and let's take the example of exactly once, is what we would do is well when we when we when we start processing, um we every time we start a Kafka transaction. Um that means we would start to um to process data, we write the changes into the changelog topic for the state, we write results into the output topic, and when we commit the transaction, committing the offsets is part of the transaction itself. Um, and that basically means the processing for this task is either successful, including recording the offsets of the input topics, or it is not. And if not, well then the Kafka transaction is doing the rollback for us automatically, because the changes into the changelog topic are aborted, the changes into the output topic are aborted, and no offsets are recorded. And when we retry basically, well, we just fetch the old offset. Uh, and we also need to do some magic to rollback the local state, but in the changelog topic and in the output topics, the rollback is done by the Kafka cluster for us. So we don't have to do anything here. That means, of course, that every task has individual fault tolerance. But you know, if one task fails, it does not affect other tasks, they just keep running, and then we just recover this one individual task. Um, while in checkpointing, if one task fails, he would say, okay, now I need to stop everything. Uh and I need to roll back everything to this globally consistent checkpoint, uh, and then I can resume processing.

SPEAKER_01

So um let me try and recap that. You you talked about committing the offset being a part of the transaction, and I want to I want to unpack that a little bit in case there's anybody who didn't follow that. So basically you've got uh Kafka streams task, and we're just going from the standpoint of one task here, um, but that task is some part of your stream processing that you're doing. You have the stream processing topology and and there's a task that's doing some of that computation. It has input, one or more input topics, and let's just say it has an output topic and potentially materializes some state in the middle. As it consumes messages, because it's a Kafka consumer at bottom, as it's consuming messages from the input topic, it's doing that within a Kafka transaction and also included so when a consumer consumes messages, eventually it has to tell the cluster basically, I've gotten so far, I've gotten to message offset, you know, 1,048,576,000 in this partition of this topic. And that consumer offset is itself a message that gets written to a topic. So and that that lets that application uh, or really, I guess the system know that stream processing application has gotten that far, certainly. It definitely made it to that offset. We've committed the fact that we we processed that message and that has been written back to the offsets topic. But the writing to the offset topic is itself done inside that transactional context. Is that what you said?

SPEAKER_00

That is correct, exactly. And and that's really the key to exactly once processing, because if you if you don't do this atomically, then you only have the choice to say, well, I I flush my results to the output topic, and afterwards I commit my offsets, or I first commit my offsets and afterwards flush the results, and then basically always gets this kind of inconsistency between at least once or at most once processing. Um and only if you do both at the same time in an atomic manner, basically in a transaction, then you get get exactly once. Got it.

SPEAKER_01

Um and I want to go back to the the complexities with checkpointing. I think basically what you said there is that a checkpoint is definitionally state, right? You have this sequence of events and a checkpoint is um I don't know, maybe I'm uh maybe I'm taking checkpoint to mean snapshot, but it's a way of saying when I got to this offset, this is what things looked like.

SPEAKER_00

Yeah, I mean snapshot is a is a totally correct description here. Basically, there's a checkpoint means the snapshot of all stateful operators, so the state of all operators being and and then what you need to add is basically also the input topic offset that corresponds to that snapshot of the state.

SPEAKER_01

Right. Okay. And the problem with that though is that it is state. Um that's that's the checkpoint with respect to this stream processor and some. Stream processor, well, definitionally, is gonna be able to have a different s a different snapshot. Is that am I because you were describing the potential pitfalls with checkpointing, and am I getting that right?

SPEAKER_00

Yeah, so the problem is really to to get us to get a checkpoint that is that it's consistent with regard to the input topic. So you cannot have like in-flight data. So let's assume you have like two stateful operators, one after each other, and you start reading uh a record from the input topic, and it updates the first and the second state. And now you say, okay, now I can actually commit to the offset of this record. But because everything is paralyzed, the processing might actually already have taken a second record and updated the first state. So if you would now say I take a snapshot, then and you take the offset of the first record, then you would basically say, Well, the snapshot now is not consistent because the first state snapshot also contains the update of the second record. And that should not be the case, obviously, because otherwise it's not consistent. Got it. Okay.

SPEAKER_01

And so that is that is that problem is essentially the reason why Kafka Streams made the engineering decision not to use checkpointing.

SPEAKER_00

I wouldn't say that. I mean it's possible to do that, uh, and other systems do this very successfully. Um they use a so-called distributed snapshot algorithm, um, where you can get consistent snapshots. Uh, for Kafka Streams, um, the main reason is really the problem where do you store the checkpoint? Because Kafka Streams is a library. And and the idea of Kafka Streams is to say, well, we we want to be lightweight, easy to use, easy to set up without other dependencies if possible. But if you do a checkpointing algorithm, then well, you need an S3, for example, or an HDFS where you actually store the snapshot. And that makes the deployment more complicated. Because if you don't have a place to store the snapshot or the checkpoint, then how do you do that? And as a matter of fact, there is already a Kafka cluster that we can use to do change logging. Why? And it doesn't add an additional dependency to the library. And so it's kind of a natural choice to go with change logging here because it makes the deployment much easier. Right. Okay, that makes sense.

SPEAKER_01

And that uh we're not really talking about the the reasons why Kafka Streams decided to be a library and not its own infrastructure, but I think we've covered those before, and again, I'm gonna link to some previous episodes. Briefly, uh I think you know you said there are systems that that use checkpointing and are very successful that way. There are also stream processing systems that are perfectly successful as their own infrastructure and and can do a lot of things very well. Um but that ends up being a poor fit uh for integration into microservices that have stream processing concerns, for example. Like what you want in that case is you want a library and nothing else. Uh and so you know Kafka Streams is is designed around that use case and very successful within that use case. But that that puts interesting constraints on the way other problems get solved and pushes you now, in this case, in the direction of changelogging and not checkpointing.

SPEAKER_00

Yeah, absolutely. And and I mean the idea to add checkpointing to Kafka Streams is floating around for multiple years already. And I'm pretty sure at some point it's gonna be implemented, but then it will always be kind of an opt-in feature. So change logging will always be our primary thing to preserve the nature of the library. And if somebody really says, well, checkpointing has a couple of nice features that I want to exploit, and I'm willing to add this dependency to my program, then it would be nice to actually have this capability to let people opt into checkpointing and replace uh changelogging with it to get the advantages of checkpointing if they want. Got it.

SPEAKER_01

And if they're willing to tolerate the like, you know, I happen to have a blob store nearby that we can use for this, or you know, if when that's when that's um not a cost, when that is a cost that they're willing to pay, then they can have it. Yeah, correctly. Someday, watch the kips. Is there an active kip on that or is it just discussion?

SPEAKER_00

No, it's just like floating around this idea. Um nobody's really working on it correctly. It came up on the mailing list a couple of times, but I think what those things are also very, very complicated to implement. And the immediate gain at the moment is it's kind of nice to have a feature, but it's not like crucial to add it. Um and I think for for many people who you know contribute on the site, and it's not their daily job to do that, uh, building a feature like that is maybe too heavyweight and too time consuming. Um, so I guess um we need to you know wait if somebody comes up who really has a lot of time spending on it and it thinks it's important enough to do. Um so so let's see. Um having said that, for KSQLDB, because KSQLDB is already like a server system, uh checkpointing might actually be a very nice feature for KSQLDB there. So maybe at some point it makes it more interesting, and somebody from the KSQLDB community uh might actually start a discussion about that. Um because there we have completely different trade-offs, uh obviously. So for if it takes the example of Confluent Cloud, well, we run the service we can easily provision like an S3 bucket and do the checkpointing for you, and the user doesn't even know about it.

SPEAKER_01

Yeah, that that is definitely a thing that Confluent Cloud is capable of automating. That makes sense. And that uh that's a great insight. I was about I was about to say, you know, yes, this is an open source project, and so somebody has to want the feature in order to invest in building it. And yes, when someone comes along comes along and wants it, there's definitely sufficient uh expertise in the community to to build checkpointing as an optional thing. But um I love the idea of case equal db being that potential client of Kafka streams uh to need that. Because yeah, that does make all the sense in the world. Because uh if you're CaseQL DB, well, there is infrastructure. You know, you have a CaseQDB server. Uh that is uh that is uh how how you run queries. And so um you know you're you're out of library land and you're in, okay, we have infrastructure here, so let's add layers. And it feels feels like less of a burden to add that that checkpointing layer. You could even do it locally on disk, right? That would that seems like that would be an option.

SPEAKER_00

Yeah, well, that's that that depends, of course. I mean, the problem is really if you do it locally on disk, um, you still would need to kind of replicate the stuff. Because if one server goes down uh and you can't access a checkpoint anymore, you you lose your fault tolerance, so that would be would be rather costly. Um that's the reason why usually people go with external systems like HDFS or S3, where you know those systems do the replication for me already and I don't have to worry about it. But yeah, it's certain it's certainly a choice. Um there are also stream processors, at least in academia, not sure if somebody implemented it, where they would basically say, well, we could do a similar thing in Kafka streams. Maybe you say, hey, I deploy my application, it's like 10 application instances. And if one instance needs to snapshot, it just stores it on two other instances and you know distribute it. Um of course then you have the problem. The problem here is really that storage on the clients then usually is more expensive. You need to provision your clients with more storage, um, because well, the state has to go somewhere. Uh, and then usually using an external system, especially if the state is large, is usually cheaper because well, you don't have like expensive SSDs or stuff like that.

SPEAKER_01

Right. Cheaper and probably slower if you're comparing, say, if just within AWS terms, your EBS volume is gonna be more per byte and faster to access than your S3 bucket, which is gonna be a lot less per byte and probably slower to access. Um but again, the building the replicated blob store amongst your KSQLDB nodes or whatever is is probably an engineering choice that teams should be encouraged not to make.

SPEAKER_00

Yeah, yeah.

SPEAKER_01

And that's that problem's been solved. Let's not solve that again.

SPEAKER_00

Exactly. And what you said before is actually quite interesting to say, you know, in in KSQLDB, we already already leave the land of library building, and we already have a server system. Uh and so checkpointing makes more sense. And that's also the reason why stream processing systems that are infrastructure usually go with checkpointing, because in their case, they kind of need to add one dependency to the system anyway to enable for tolerance. And this dependency could be a Kafka where you do the change logging, or it could be an HDFS or an S3 where you do the checkpointing. Um, and so it doesn't really matter which one you choose from a complexity point of view, you need this external dependency. Um, and then checkpointing has a couple of nice features, those people want to exploit, and so they natively go with the checkpointing approach. Um, because from them, for from their point of view, the trade-off to go with checkpointing uh is the better one.

SPEAKER_01

Of course. And just to be specific, say if you're flink, um you're gonna introduce the dependency, the storage dependency that enables checkpointing. Because while in practice, if you're flink, yes, we all know there's a Kafka near you. Um like that's that's probably you know two or three sigmas of your your Flink systems if there's a Kafka deployment. But maybe you don't want to introduce Kafka as a literal dependency when you could just make sort of a some sort of pluggable blob storage API that we all know is probably S3 or Google Cloud Storage or whatever, most of the time. Uh that that's that does seem like the more responsible dependency to introduce uh once you're committed to being infrastructure.

SPEAKER_00

Yeah, yeah, absolutely. And especially if you go back like like five to ten years where MapReduce was big. You know, most people already had an HDFS. That's kind of, well, if you run Flink, you know, just took it up with your HDFS and you're good.

SPEAKER_01

Yeah. Just put it in HDFS, it's fine. It's gonna be there. I mean, that's that's what that's what it was that that was the storage underlying the stream processor, so go ahead and use it. Um but no, that is actually really helpful to me to see that. That changelogging, Kafka streams knows it has a dependency on Kafka. It is a Kafka thing. There is uh there will be Kafka topics accessible to you. So do change logging. Uh checkpointing nice, sure, it does some nice things at the cost of a dependency that we would rather not build in until we want it.

SPEAKER_00

Yeah, that's that's a totally, totally fair summary.

SPEAKER_01

Now tell us about hot standbys. Uh I know uh I'll just frame this a little bit. Kafka Streams is a horizontally scalable library. Uh basically, I'm if I have a Kafka Streams application, I am a consumer group. Therefore, I can scale horizontally. I can deploy two or five or ten instances of my application, and the consumer group rebalance protocol will assign partitions in my source topic to consumers in the group, you know, those nodes in my Kafka Streams cluster. With that picture in your head, everybody, Matthias, what are hot standbys?

SPEAKER_00

Yeah, um, so maybe before I answer that, um let's go go back to checkpointing versus change logging. Because you mentioned before that that there are different like advantages of the of the of the of the two techniques. And as you mentioned already, you know, if you if you write a checkpoint somewhere into a blob storage and you want to recover the checkpoint, this is usually actually pretty pretty fast. And usually it's actually faster than replaying a changelog topic. So then Kafka streams, when you have state and something goes down and a rebalance happens and the task is migrated to a different instance while there's no state. So the first thing the task is doing is it's saying, well, what is the changelog topic of the state? It goes to the changelog topic and it has to read the whole topic. So of course, for key value stores, those topics are log compacted. So depending on your on your configuration, if your log compaction is efficient enough, the size of the topic corresponds to the size of the state. So ideally you have exactly one record per key in the changelog topic. Um but but reading this from a Kafka topic is compared to just copying a file of a checkpoint uh relatively slow. Um so now the point is, and that now we're going to the land of high availability, that so check change logging makes Kafka streams for tolerance, but it doesn't make it highly available. Um for checkpointing, it's kind of borderline if you want to call it high available, it depends on the size of the state, especially if you have a very large state in the in the terabytes, then it can also take five to ten minutes to load a complete checkpoint. Um and now standby tasks come into play. And the idea is to say, well, we we want to have a technique that allows us to be highly available. Uh and how do we do this? Well, the idea is to just say, well, we have we have one task that is doing the actual processing and updating the state. And then we're going to want to have like a hot standby shadow copy of this state uh somewhere else. And now if if the task actually has a failure, um we don't need to recover the state from the changelog topic because we already have a hot standby ready. So we basically fail over to this hot standby and can resume processing immediately because the state is already there.

SPEAKER_01

So um how do you how do you control that? Because I I um I'm just thinking about how partitions get assigned, um, which more or less gets done for you, right? There's the consumer group rebalance protocol that that makes that happen. Uh and streams tasks are you know, in a sense, running on top of that. So if I deploy more nodes, I guess what uh what makes a node a hot standby? Do I necessarily have to have more nodes than I have partitions? How do I make sure the partitions, like all of my partitions, have a hot standby? Um there's almost an operational question, but uh how do you do that?

SPEAKER_00

Yeah, so for for the user itself, it's just a configuration. You just say, hey, give me a hot standby. Um and that basically means that for for every state, it's basically kind of a replication factor in a topic. You say replication factor of a topic by default is one. If you set it to three, then every topic partition gets two shadow copies, the followers. And in Kafka streams, it's the same thing. By default, standby replicas are set to zero, so we don't count the active task. And if you sets this configuration to one or to two, then Kafka Streams takes care to deploy those additional hot standbys for you. Very nice. What you need to know is for the for the rebalance protocol, um, so the rebalance protocol, so the brokers basically give you, you could say, some basic infrastructure to do rebalancing, but rebalancing itself is completely controlled by the client. And so in fact, Kafka Streams does not use the regular uh standard implementation of the rebalance protocol of the consumer, but it replaces it with a custom implementation that is tailored to Kafka Streams. Um that allows us to say, well, if we have an active task on host A and we have standby configured, then we make sure that we assign the standby task onto host B, so on a different node. Um, because we have to have the knowledge about that. Um so what you said basically is correct. In order to make hot standbys useful, you need to at least like two application instances, because then you can always use the other one to host the hot standby. Got it.

SPEAKER_01

And so turning on that, and then that's why Streams has to do its own its own rebalance protocol is uh because hot standbys have to be aware of of basically have to be uh aware of the activities that layer down in the stack. And so they they pull that up. Streams pulls that layer of the stack up into itself.

SPEAKER_00

Yeah, exactly. I mean that's only one reason. There are multiple other reasons why we need to control the rebalancing and the partition assignment. Um but but that is very crucial, yeah. And the point is also Kafka Streams here is very, very resilient. Um, let's say you you configure standby tasks two and you only deploy one instance, we would still start up and running, and we would just log your warning and say, hey, we couldn't deploy your standby tasks because there's no capacity where we could place them. Because placing the standby tasks on the active task node does, of course, obviously not really make sense.

SPEAKER_01

Uh no, it doesn't. So uh good. It'll give you a warning if it can't place them. And from a resource utilization perspective, like that's a that that figures into capacity planning. You will have to deploy such instances that you can have you know, you you can get your uh stream processing work done uh without under consuming with the latency that you want to reach, and also have hot standbys. Yeah, exactly. But that gets you failover times that probably are unbeatable.

SPEAKER_00

Yeah, if the if the hot standby can can uh is really hot, um then usually I mean you need already like 15 seconds, or I think 10 seconds actually is the default session timeout to detect a failure. Um and then this 10 seconds are usually enough for the hot standby to even catch up to the latest offset. Um and then you have have instance failover during rebalancing.

SPEAKER_01

Right. When those are all things, things that are going to happen regardless of whether you've got hot standbys.

SPEAKER_00

Yeah, yeah, exactly. Of course, there's always you know got shares and you need to tune your system accordingly. Um so that's also a reason why if if you have hot standbys, um, you should kind of kind of monitor them um to some extent to make sure they are really hot and not that those consumers are lagging. Because if they're lagging, then of course you you use the instant failover. It would still work and it would still speed up your failover time, obviously, because let's say you have a standby that is only caught up to 80%. Well, you still save 80% of the time to read the whole changelog topic. Uh, you only need to replace the last 20% until you can consume uh start processing again. But yeah, uh in general, you you want to monitor it. Right.

SPEAKER_01

All right, next question. Uh I'm learning a lot, everyone else. I hope you are too. Um, watermarking. I hear about watermarking in stream processing. And uh what is watermarking and why does Kafka streams not do it?

SPEAKER_00

Yeah, so the watermarking is is a very interesting thing. It's kind of related a little bit to checkpointing, because the way you implement watermarking and the way you implement checkpointing is basically the same base algorithm. Um and in the end, watermarking is a way to track time. Um so when you you do stream processing, and let's say you have you have you have a you have you have a you have a windowed operation, um, then you always need to deal with out-of-order data. And so a watermark um is basically a time marker in the stream. Um and so watermarks they usually could say uh a low time watermark, um, where it says, well, if if the watermarks that carries a timestamp arrives at an operator, uh and the watermark carries timestamp 100, then it promises to the operator that no record with a timestamp lower than 100 will arrive in the future. Uh, and that then allows the operator to say, well, if I have a time window from 10 to 15 and from 15 to 20 and so forth, the watermark is 100. Now I can close all those windows because I know no new data will arrive, all data have been seen that belongs to this window. So this is basically just just one mechanism to track time. Uh, and and Kafka Streams doesn't do it mainly because of like architectural limitations, you could say, because those watermarks need to flow through the whole data flow program in a consistent manner. So usually the source nodes are starting to inject those watermarks in a in a kind of consistent manner. Um, if you go back to Flink, there is a there's a job manager. Um, it's a central component, and the job manager says, okay, it's time to send a watermark. It informs all sources that a watermark needs to be injected. Um those sources are doing it, and then the watermarks are propagated in a consistent manner through the whole data flow program. Why can't Kafka Streams do this? Because we basically rely on repartition topics to shuffle data around. We don't have in memory channels. Um that means when one task is communicating with another to a topic, the only way of communication is a repartition topic. Um That makes it hard to actually propagate a watermark downstream because the only way we could potentially do this is to write a message into the repetition topic. But the repetition topic actually is supposed to hold data. So now you basically have this problem that the repetition topic needs to be able to also store metadata. And that is hard to do.

SPEAKER_01

Yes. And so the basic problem Watermark is trying to solve is to tell you when you can be sure an old window has closed. Or you don't need to worry about late arriving data that you're going to have to go put into a window that has closed. Correct?

SPEAKER_00

Yeah, correct. So that means to be a little bit more precise in Kafka streams or also in Flink. There's a distinction between out-of-order data and late arriving data.

SPEAKER_01

Yes.

SPEAKER_00

And late arriving data is really data that comes after the watermark. So basically violates the contract and then is not processed anymore.

SPEAKER_01

Got it. Out-of-order data is data where the message timestamp is different from the sequence in the log. Correct.

SPEAKER_00

Yeah.

SPEAKER_01

Right.

SPEAKER_00

And then when the when the watermark comes, it tells you, you know, no record with timestamp 100, for example, will come. But it's after the watermark, now record comes as timestamp 90. It's out of order, but it's also late because it's after the watermark. And then usually the corresponding state of this window is already perched. So you can't process this record anymore and you drop it on the floor. Gotcha.

SPEAKER_01

Gotcha. And that's that's a a category of records. I mean, records will arrive that that um are simply too late.

SPEAKER_00

Yeah, that could happen, exactly.

SPEAKER_01

Yeah. Uh so what does and it absolutely makes sense why uh Streams doesn't want to do that without that central infrastructure component uh to to uh I guess conduct when watermarks happen, you know, like a like an orchestra conductor. So what does it do? Obviously it it robustly supports windowed operations and out-of-order messages, not late arriving, but out-of-order data. Uh so how does it get away with that?

SPEAKER_00

Yeah, so Kafka Streams basically uh implements a mechanism, what you could could call, if you want to, like a high watermark approach, um, in contrast to a low watermark approach, what actual watermarking is. So what we basically do is we we observe the progress in stream time. So every record carries a timestamp. And so what we do is we basically say, well, we just record the maximum timestamp we have seen per partition. So we read data from an input topic partition, records coming in one, two, three, four, five, six, seven, and so forth. We just remember this is the largest one we have seen. Um, and then for for a windowed operator, the operator itself has a configuration, what we call its grace period. Um and so the grace period tells us how long the window should stay open with regard to this, what we call stream time progress. So let's say you have a window of size 10, and the window goes from 100 to 110 as an example. So now we see data with timestamps, you know, 100, 191, 192, and so forth, it all falls into this window. Um and now we see the records with timestamp 111, 12 that are after the window, but they might still be out-of-order data. So the window is still kept open. Um, and then when let's say in this case we have a grace period of 10 seconds, when we see the first record with timestamp 120, what is basically the end of the window, so this 110, plus this 10 seconds of grace period, then we say, okay, now we assume that we have seen all out-of-order data and we close the window for you. So it's kind of more like a local decision on a per operator basis based on the input data we are seeing to reason about time progress compared to to what a watermark approach does. Um so for basically it's kind of the same idea because in the end, what we're basically doing, you could think of kind of you know have local watermarks because the operator just observes the time progress and at some point makes a call to do a cutoff point. And so when we when we basically reach this 120 in stream time, we just close the window. And that is the same thing as if Flink would say at this point in time, I send you a watermark with timestamp 110 that promise you that this kind of you know um 110 is already closed.

SPEAKER_01

We've talked a little bit about the framework library distinction and how Kafka Streams is differentiated by being a library. Uh that's a Java library, it's a dependency, and it's an API you you code against and it it is deployed as a part of your application. I hinted at some of the reasons why I think that's a good idea, but I guess even without advocacy, you know, I I tend to always want to make the case for Kafka Streams. But um What are the trade-offs? And it's obvious that they're trade-offs, right? They're they're they're both good ways to build a system, but um, what are the utilities that Kafka Streams was trying to maximize when it decided to be a library and not infrastructure?

SPEAKER_00

Yeah, I think so. The main purpose was always being super, super easy to use and to deploy, and to not depend on other systems because that's really making it easy to build real applications, what what you want to do. So if you have this kind of infrastructure systems, then very often the application that you try to build is basically split into two parts. You have the stream processing part, computation that you have to package up and send into a processing cluster. But then you also have your actual applications that is going to use as a result of this data. And in Kafka streams, you just build a Java application to do both. And so you don't have this kind of separation of things that are actually natively belong to each other.

SPEAKER_01

Agreed. That sounds like uh, again, there's there's room in the world for more than one way to do this, but uh given that distributed systems infrastructure is difficult to deploy and you'd rather not deploy it, um this seems like a this seems like a uh a good trade-off to make if application development is uh your goal and not say ingesting the Twitter firehose. Because it turns out you only you only want to stream process the Twitter firehose so often.

SPEAKER_00

Yeah, yeah, totally. And of course there's uh there's a lot of stream processing jobs out there where it's really about data crunching, you could say. You say, you know, I have this input stream, I don't really build an application, I just have the stream processing job that is, you know, massaging my data, you know, cleaning it, whatever, doing some results and then push it somewhere else, and there's no real application involved, then of course it doesn't really matter what you do. You can use either uh infrastructure system, especially if you have a team that deploys it for you, or you use a library to do it. It doesn't really matter. But if you build an application that is actually not doing something else, then the infrastructure thing is a little bit more complicated.

SPEAKER_01

Yes. Um I feel like we've barely scratched the surface of this. There's a lot more to talk about. We're kind of at time, but I think I would like to have you back on the show to continue this deep dive in the future, uh, because it's important stuff. It's important to be able to understand the trade-offs and the comparisons with other kinds of stream processing systems. And honestly, even if you don't care about those trade-offs at all, if like, well, you know, you're a Kafka Streams developer and that's just kind of how it is, uh, it's always good to learn a little bit about the the decisions that informed the framework and some of the concepts behind them. So Matthias, thank you. And will you come back on?

SPEAKER_00

I would love to, opposite Lee. I mean, you know that I also like to give talks at Kafka Summit to educate people about internals of Kafka streams and being on the podcast is a very good other way to do that. So uh yeah, absolutely.

SPEAKER_01

Excellent. My guest today has been Matthias Sachs. Matthias, thanks for being a part of Streaming Audio.

SPEAKER_00

Thanks for having me. My pleasure.

SPEAKER_01

Hey, you know what you get for listening to the end? Some free Confluent Cloud. Use the promo code 60PDCAST. That's 60PDTAST to get an additional $60 of free Confluent Cloud usage. Be sure to activate it by December 31st, 2021, and use it within 90 days after activation. And any unused promo value on the expiration date will be forfeit, and there are a limited number of codes available, so don't miss out. Anyway, as always, I hope this podcast was helpful to you. If you want to discuss it or ask a question, you can always reach out to me at TL Burgland on Twitter. That's T-L-P-E-R-G-L-U-N-D. Or you can leave a comment on a YouTube video or reach out in our community Slack. There's a Slack sign-up link in the show notes if you'd like to join. And while you're at it, please subscribe to our YouTube channel and to this podcast wherever fine podcasts are sold. And if you subscribe through Apple Podcasts, be sure to leave us a review there. That helps other people discover us, which we think is a good thing. So thanks for your support, and we'll see you next time.

Head Shot

Tim Berglund

Host
Head Shot

Adi Polak

Co-host
Head Shot

Viktor Gamov

Co-host