Kafka Streams and Protobuf
CC Attribution 2.0 Belgium:
You are free to use, adapt and copy, distribute and transmit the work or content in adapted or unchanged form for any legal purpose as long as the work is attributed to the author in the manner specified by the author or licensor.
Valiente, Clemens
A lot of components in the Kafka and hadoop ecosystem assume you are workingwith avro messages. In this talk I will highlight some of the advantages andchallenges of stream processing with protobuf encoded messages, and why itmight be worth using them. A powerful feature of protobuf messages is its handling of unknown fields, andthe ability to partially parse messages with different schema versions or noschema at all. This helps in stream processing when you don't actually need toknow all message fields in order to process and forward them.
 So a very quick introduction into Kafka, then some concepts about Kafka streams, introduction on Google protocol buffers, and what kind of stream processing you can do with that. All right, click Overview. Apache Kafka is a message queue system. Nowadays, it's evolved into a distributed streaming platform. So you can put messages in several queues. Their queues are called topic in the... Sorry, yeah. All right, so Kafka is a distributed streaming platform. You have message queues, several, which are called topics. Each topic is partitioned and replicated, so it allows you to scale out by putting more servers in your cluster and putting more partitions in your topic. And each topic is... Each partition is usually replicated. So if a machine goes down, you have the data also somewhere else on a different machine. And one important part, especially when it comes to Kafka streams, is all of the consumer group logic. So you can have several consumers forming a consumer group and reading from a particular message queue. And this allows you also to scale the reading process, and it works also quite similar. So in this case, we have two different consumer groups which are doing different things each, but within each group, you have several threads. And basically, they coordinate with each other to distribute the different partitions among each other. So if one of your consumers in that group goes down, the remaining will take over the work that the consumer was doing. And if you add more consumers, then again, the consumers will coordinate with each other to distribute the workload again, which allows you also to easily scale the reading, the consuming from your Kafka cluster. All right, so what is Kafka Stream? So there have been a lot of stream processing frameworks out there, and sometimes you might wonder why is there need for another one? Why is... What... So, one very important feature is that it's a really small library. It's not a huge framework that you get with Spark. It doesn't have any further external dependencies, so when you write a stream processing up in Kafka Streams, you just put a dependency on Kafka Streams in your cradle, and that's it. And in the end, what you get is a very simple small Java application, usually less than 10 megabytes. And you can run that one everywhere, so you can put it on Yarn, on MISOS, Docker, Kubernetes, all that. You can also just copy the char on a server somewhere and start it there. So it's pretty agnostic on how you want to deploy it, while most other stream processing frameworks, they really have very explicit requirements on where you deploy them. Today, the Kafka Streams basically uses this Kafka consumer group logic that I introduced for scaling and fault tolerance. So other stream processing frameworks had to do really invest a lot of work to get elastic scaling and all that working. And Kafka Streams basically got it for free by using the consumer logic, and that consumer logic with Kafka has been already tested in BattleHard for many years by now, so it's also something that by now has proven to be working. And due to its very small footprint, it's ideal for a microservice architecture, so you can have your different services all run Kafka Streams applications that basically to the transformations that read from the Kafka cluster and then expose that information back as a Kafka cluster, as a topic in the Kafka cluster. And Kafka Streams has two different APIs, both in Java at the moment, high-level DSL, which is very similar what you would get in Spark with those filter join and map functions that you can use, and a low-level processor API, which is more explicit and at the moment also a bit more powerful. It's similar to what you get in Storm and SAMHSA. It also has the concept of tables and streams that are separate, so streams are basically infinite event logs, so you have all these messages coming in, and the assumption is that it goes on forever. And tables are finite with updates, deletes inserts on the primary key, and both can be represented as message queues, as Kafka topics, by just taking the change log of a table. So we have that here, if a table and each action that modifies its content is a change log entry in a stream, and this allows you to basically have stuff, for example, static lookup tables that you join with incoming data streams. So one classic example is that you have customer events coming in, and you want to look up customer information and enrich that. Kafka streams, of course, also allows stateful processing, and it stores its state locally in a ROX DB by default. You can also plug in any other DB if you look like. Oh, doesn't skip. Yeah. All right, so you have usually a ROX DB as a local state, and this state, this ROX DB is a key value store. It's again some kind of a table, which, as we saw before, can be seen as a change log, and a change log is a stream, so the way this state store is made available is that it's written back to Kafka as a topic, as a change log topic for your state store, and this is the way that Kafka streams, handles, rebalancing, and fault tolerance, because any other instance of your application can just read the store of your, the state of your streaming application from that change log and recover it. So instead of these check pointing or lineage that you get in other streaming application, you really can recover the full state of your application in case it goes down, or in case you want to launch additional instances of it. So this allows you to really scale your streaming applications up very seamlessly. You just launch another application on a different computer or different server, for example. It connects to the consumer group. It reads from the change log, recover, repels the local state, and takes over. Nowadays, in the new version, you can actually even query the local state. So depending on your use case, you don't even need to worry, okay, now my streaming application writes the data back to Kafka, and then from Kafka, I need to write it to some other kind of database, and from that database, my application can finally read it. Now with this way, you can actually query the state of your streaming application directly. You don't need additional infrastructure in between, which again allows you to build very lightweight applications that don't need a lot of additional infrastructure. A lot of tools and components are on the Confluent platform, and Kafka make the implicit assumption that you're primarily using Avro. So tools like Kafka Connect, which is responsible for connecting Kafka to other systems like relational databases or Hadoop, and they usually work with Avro messages, same with the schema registry, which at the moment only works for Avro messages. So Avro usually is the natural choice if you start using Kafka. It doesn't require a lot of effort to implement since you already get everything. It's also a great choice and solution. But one downside is that you always need to have the correct schema the message was written in in order to be able to read that message reliably. And if you don't have the schema available to you for some reason, then all your data is just byte garbage. You can't really do anything with it. And this is where Google Protocol buffers come in. It's very similar to Avro in a lot of features. It's a binary message. It has a defined schema. It has a support for lots of languages to read and write protobuf messages. But a big difference is that you can also read these protobuf messages with a different version of the schema or even with no schema at all, you can get quite some information out of it. Well for as an example, if you're one of the tutorial protobuf messages, which is a person and this person message has several fields, a name, an ID, as a required fields and an email as a string and then four numbers as an embedded message. So you can also have basically hierarchies of protobuf messages. And each field basically consists of four parts. In denifier, if it's required, optional or repeated, the type like string or integer, field name and then a field ID. So this is one example person, John Doe. You can see here with full schema, we get all the information. We have the name, the ID, email and phone number. And as a binary message, it looks like this. So you can still see all the strings in that there, but everything else is just bytes that you can't as a human really read. But now we can also read this message back with a different schema version. So let's say we have an old schema that doesn't know about the email field yet. We still get all the fields that we know about, the name, the ID and the phone. But also additionally, we get this field that we don't know anything about, the email. And there we just get a field ID and not the field name. But we can still even just by looking at it kind of guess that this is supposed to be an email. So even though we haven't even heard of this field before, we can still get the information out of there and read it. And if we try to read the message here without any schema at all, this is what we're going to get. We still have all the fields in the same order. We just lose the field names instead of only the IDs of the field. So even with knowing nothing at all about the message, we can still recover quite a lot of information. This is basically the output of a protocy decode raw. You can put any protobuf message in there and it will tell you what fields are in there. And this is the kind of thing that would be very hard to do with AVROW. All right. Another very cool feature are unknown fields. So these are the fields that we don't know anything about. And not only can you read them, as I showed you with your schema, you can also pass them on. So if you read a message and you've created a new one from this one, then all the unknown fields that you can find in your original message will be copied over to the new one, which allows you to basically do things like decorations or extracting common fields or envelopes without needing to know the schema information of your original message. You just put that message inside another message. You will take over all the fields and send it on. And you don't have to worry about breaking or destroying the original content. It will still be fully there. So now how can you use those two components together for stream processing architecture? So generally, you still have all the usual message producers like logs, databases, APIs, and sources, they all write their information in protobuf to your Kafka cluster, where you have streaming processors that enrich, showing, filter, aggregate all the usual stuff with your topics and create new topics with the output. And that one can go to your various consumers, for example, databases, reports, dashboards, applications, machine learning, or storing it in Hadoop, for example. And you can exchange schema versions through, for example, Git repository or some other central place where everyone can retrieve these definitions and the developers and engineers that write the message, they write the input data, they provide their protobuf definitions, the engineers writing the streaming processors, they also provide the definitions for their output while reading the one from the engineers and the consumers, they just need to read the schema definitions to be able to parse everything correctly. So how would that look like in an example? Here we have three different systems that emit events to Kafka. So for example, a web service that writes log events to your stream. So log events from your website, for example. In there we have a timestamp field, a session ID and a type, like, okay, use a looked at a product or a customer put a product in the shopping cart. And then we have the optional fields, customer ID and product ID. Then from your customer database, you get these customer messages, which again have a customer ID, a name and an email. And your product database provides the product messages. And with a product ID, a name and a price. These then you have a stream application joining these three topics and emitting a new message, which is a very simple message. My rich event message basically has three fields, one event field, one customer field and one product field. So I actually don't need to write a lot of schema definitions because I can just import the previous ones there. And the output of this streaming application is again a stream in my Kafka cluster of these rich events. And then in that I can, for example, plug in another application that does something like sessionization, so it takes the session ID out of there and groups by them. And my session message is also again very simple. Session message just consists of repeated rich event messages. So makes it very easy to define these things. Of course, at this step, I can also do further aggregations like include fields for counting how many different products a user has looked at or how many interactions they did with our website and so on. But it's basically a very simple version. And that one is then something that you can, for example, read by machine learning algorithms for personalization or recommendation engines. You can put it in reports or you can let your analysts read it and do analysts and do their talk reports on it. Now going back to this chart, so the skid repository or whatever you're using, I mean, it's basically just like the Avaro schema registry that comes with a confluent platform of Kafka. So why is this one, what's the advantage of this one? Now it can always happen that not all the protobuf definitions there are actually up to date, for example, the developer could have forgotten to commit their message or the host of your git repository accidentally deleted their production database and is now trying to recover a backup. So it could always happen that this one doesn't quite work and someone doesn't have all the schema versions that they need. So what happens in that case? Let's, for example, assume that in your product database, someone adds a new field for the color of your product and starts writing that to the Kafka product stream. Now your streaming application is starting to, or is processing these events and putting them in the rich event message. And all that happens is now that the product contains an unknown field, which is the color, but everything else will just keep on working. And even further down, it causes no trouble at all. My session will still contain these rich events and these rich events all will contain a product with unknown fields. And even at the end, the applications can still read these messages. And if the BI analyst or data scientist at the end looks at the messages, he can find that unknown field. He can look at the value and might even be able to find, to determine that it's supposed to be color. Maybe he was even the one that wrote the task to the engineer in the first place to put the field in there. So the whole pipeline basically still works the same way. Similar if you just delete a field like here, the price. For a protobuf, there's no difference between a field that simply doesn't exist and an empty field with a null value. They both are simply not included in the protobuf message. So all you're going to have now is a product that contains a null price. And since your price since the very beginning was marked as an optional, hopefully the engineers writing the stream applications kind of put in logic to handle null prices already. So your streaming applications will continue running uninterrupted. So in general, what does this allow you? So it means that the different teams that are responsible for data, for producing the data, for processing the data, for consuming the data, they can all move at their own speed. There's no real strict alignment for releases necessary. So if you have a traditional OLTP system that feeds into a data warehouse and you want to do schema changes across the system with several, for example, MySQL servers involved, it's very, very painful because you need to make sure that each of the queries takes that into account. But here we can really do a hands-off in data engineering. The people in the middle writing the streaming process, they don't need to do anything. They just read the data and write it to the output. And everyone can basically upgrade to new schema versions at their own speed. And only the actual producers and consumers need to align on including new information in your messages and the pipeline in between just forwards them without needing to know about it. Of course, not everything is great about Protobuf. There are also some downsides compared to Avro, so there's no... Why do you kind of need to hand-craft your schema and be very considerate about it so you cannot reuse your field IDs? That's the important thing. And you really need to stick to them. And you need to also think about what you may mark as required and optional because once something is required, you basically can never change it. There's also less implementations for it around Kafka and Hadoop ecosystem and less people using it. So you will find out that you will often have to write the serializer, the serializer yourself. Therefore, Avro, you basically find enough implementations of that. A big issue is that Google kind of wants to remove these unknown fields in Protobuf free. It's a very long discussion now going on in the related issue on GitHub. And one more thing is that Protobuf messages usually are one or two bytes larger than Avro messages. Nowadays, I don't think that's a really big deal, but for some people that is an important factor. All right, thank you. I hope you learned something about Kafka streams and Protobuf and maybe have some inspiration yourself. Do you have any comments or questions? Yes? I have a question about navigation. I have two servers running after that. And they get each other. I don't need to send them out or make them. Are there any way to make sure that all messages that we give to those servers? Yeah, I mean, if you have a... So the question was, if I only have two servers in my Kafka cluster and one of them goes down, how can I make sure that I don't lose any data and things still keep on running? Well, if you set your topics to all have a replication factor of two, it means each of your partitions, everything should be existing on both servers. So in that case, basically, they would be identical. They both would have the full data set. And if one of them goes down, the other one still has everything. And the producers would then all go to that one running Kafka server and write everything there. And the consumers would also start only reading from that one. Did that answer the question? Yeah. One producer, I'll just say message. Yeah. Either any delay between the message reach the secondary side or...? I mean, usually the producer sends out the Kafka cluster, sends out an ACK signal back to the producer that it received the message and processed it. And you can also define when it does that. So you can say only acknowledge a message if it has been replicated on both machines, so you can make sure that nothing is going missing. That's actually a talk by Gwin Shapira about when messages absolutely have to be there. If you want to have a very strong guarantee that your messages are there, recommend looking that up. It's very helpful in that topic. Yeah. Yeah. Yeah. I have a question, extension of this question actually. So if you have Kafka clusters, I mean, two data centers. Yeah. I'm actually after you. With that work, you mean these two data centers and the other one, you mean the one that you have to be able to do? Yeah. Yeah. Yeah. Yeah. Yeah. Yeah. Yeah.
 So a very quick introduction into Kafka, then some concepts about Kafka streams, introduction  on Google protocol buffers, and what kind of stream processing you can do with that.  All right, click Overview.  Apache Kafka is a message queue system.  Nowadays, it's evolved into a distributed streaming platform.  So you can put messages in several queues.  Their queues are called topic in the...  Sorry, yeah.  All right, so Kafka is a distributed streaming platform.  You have message queues, several, which are called topics.  Each topic is partitioned and replicated, so it allows you to scale out by putting more  servers in your cluster and putting more partitions in your topic.  And each topic is...  Each partition is usually replicated.  So if a machine goes down, you have the data also somewhere else on a different machine.  And one important part, especially when it comes to Kafka streams, is all of the consumer  group logic.  So you can have several consumers forming a consumer group and reading from a particular  message queue.  And this allows you also to scale the reading process, and it works also quite similar.  So in this case, we have two different consumer groups which are doing different things each,  but within each group, you have several threads.  And basically, they coordinate with each other to distribute the different partitions  among each other.  So if one of your consumers in that group goes down, the remaining will take over the  work that the consumer was doing.  And if you add more consumers, then again, the consumers will coordinate with each other  to distribute the workload again, which allows you also to easily scale the reading, the  consuming from your Kafka cluster.  All right, so what is Kafka Stream?  So there have been a lot of stream processing frameworks out there, and sometimes you might  wonder why is there need for another one?  Why is...  What...  So, one very important feature is that it's a really small library.  It's not a huge framework that you get with Spark.  It doesn't have any further external dependencies, so when you write a stream processing up in  Kafka Streams, you just put a dependency on Kafka Streams in your cradle, and that's it.  And in the end, what you get is a very simple small Java application, usually less than  10 megabytes.  And you can run that one everywhere, so you can put it on Yarn, on MISOS, Docker, Kubernetes,  all that.  You can also just copy the char on a server somewhere and start it there.  So it's pretty agnostic on how you want to deploy it, while most other stream processing  frameworks, they really have very explicit requirements on where you deploy them.  Today, the Kafka Streams basically uses this Kafka consumer group logic that I introduced  for scaling and fault tolerance.  So other stream processing frameworks had to do really invest a lot of work to get elastic  scaling and all that working.  And Kafka Streams basically got it for free by using the consumer logic, and that consumer  logic with Kafka has been already tested in BattleHard for many years by now, so it's  also something that by now has proven to be working.  And due to its very small footprint, it's ideal for a microservice architecture, so  you can have your different services all run Kafka Streams applications that basically  to the transformations that read from the Kafka cluster and then expose that information  back as a Kafka cluster, as a topic in the Kafka cluster.  And Kafka Streams has two different APIs, both in Java at the moment, high-level DSL, which  is very similar what you would get in Spark with those filter join and map functions that  you can use, and a low-level processor API, which is more explicit and at the moment also  a bit more powerful.  It's similar to what you get in Storm and SAMHSA.  It also has the concept of tables and streams that are separate, so streams are basically  infinite event logs, so you have all these messages coming in, and the assumption is  that it goes on forever.  And tables are finite with updates, deletes inserts on the primary key, and both can be  represented as message queues, as Kafka topics, by just taking the change log of a table.  So we have that here, if a table and each action that modifies its content is a change  log entry in a stream, and this allows you to basically have stuff, for example, static  lookup tables that you join with incoming data streams.  So one classic example is that you have customer events coming in, and you want to look up  customer information and enrich that.  Kafka streams, of course, also allows stateful processing, and it stores its state locally  in a ROX DB by default.  You can also plug in any other DB if you look like.  Oh, doesn't skip.  Yeah.  All right, so you have usually a ROX DB as a local state, and this state, this ROX DB  is a key value store.  It's again some kind of a table, which, as we saw before, can be seen as a change log,  and a change log is a stream, so the way this state store is made available is that it's  written back to Kafka as a topic, as a change log topic for your state store, and this is  the way that Kafka streams, handles, rebalancing, and fault tolerance, because any other instance  of your application can just read the store of your, the state of your streaming application  from that change log and recover it.  So instead of these check pointing or lineage that you get in other streaming application,  you really can recover the full state of your application in case it goes down, or in case  you want to launch additional instances of it.  So this allows you to really scale your streaming applications up very seamlessly.  You just launch another application on a different computer or different server, for example.  It connects to the consumer group.  It reads from the change log, recover, repels the local state, and takes over.  Nowadays, in the new version, you can actually even query the local state.  So depending on your use case, you don't even need to worry, okay, now my streaming application  writes the data back to Kafka, and then from Kafka, I need to write it to some other kind  of database, and from that database, my application can finally read it.  Now with this way, you can actually query the state of your streaming application directly.  You don't need additional infrastructure in between, which again allows you to build very  lightweight applications that don't need a lot of additional infrastructure.  A lot of tools and components are on the Confluent platform, and Kafka make the implicit assumption  that you're primarily using Avro.  So tools like Kafka Connect, which is responsible for connecting Kafka to other systems like  relational databases or Hadoop, and they usually work with Avro messages, same with the schema  registry, which at the moment only works for Avro messages.  So Avro usually is the natural choice if you start using Kafka.  It doesn't require a lot of effort to implement since you already get everything.  It's also a great choice and solution.  But one downside is that you always need to have the correct schema the message was written  in in order to be able to read that message reliably.  And if you don't have the schema available to you for some reason, then all your data  is just byte garbage.  You can't really do anything with it.  And this is where Google Protocol buffers come in.  It's very similar to Avro in a lot of features.  It's a binary message.  It has a defined schema.  It has a support for lots of languages to read and write protobuf messages.  But a big difference is that you can also read these protobuf messages with a different  version of the schema or even with no schema at all, you can get quite some information  out of it.  Well for as an example, if you're one of the tutorial protobuf messages, which is a person  and this person message has several fields, a name, an ID, as a required fields and an  email as a string and then four numbers as an embedded message.  So you can also have basically hierarchies of protobuf messages.  And each field basically consists of four parts.  In denifier, if it's required, optional or repeated, the type like string or integer,  field name and then a field ID.  So this is one example person, John Doe.  You can see here with full schema, we get all the information.  We have the name, the ID, email and phone number.  And as a binary message, it looks like this.  So you can still see all the strings in that there, but everything else is just bytes that  you can't as a human really read.  But now we can also read this message back with a different schema version.  So let's say we have an old schema that doesn't know about the email field yet.  We still get all the fields that we know about, the name, the ID and the phone.  But also additionally, we get this field that we don't know anything about, the email.  And there we just get a field ID and not the field name.  But we can still even just by looking at it kind of guess that this is supposed to be  an email.  So even though we haven't even heard of this field before, we can still get the information  out of there and read it.  And if we try to read the message here without any schema at all, this is what we're going  to get.  We still have all the fields in the same order.  We just lose the field names instead of only the IDs of the field.  So even with knowing nothing at all about the message, we can still recover quite a  lot of information.  This is basically the output of a protocy decode raw.  You can put any protobuf message in there and it will tell you what fields are in there.  And this is the kind of thing that would be very hard to do with AVROW.  All right.  Another very cool feature are unknown fields.  So these are the fields that we don't know anything about.  And not only can you read them, as I showed you with your schema, you can also pass them  on.  So if you read a message and you've created a new one from this one, then all the unknown  fields that you can find in your original message will be copied over to the new one,  which allows you to basically do things like decorations or extracting common fields or  envelopes without needing to know the schema information of your original message.  You just put that message inside another message.  You will take over all the fields and send it on.  And you don't have to worry about breaking or destroying the original content.  It will still be fully there.  So now how can you use those two components together for stream processing architecture?  So generally, you still have all the usual message producers like logs, databases, APIs,  and sources, they all write their information in protobuf to your Kafka cluster, where  you have streaming processors that enrich, showing, filter, aggregate all the usual stuff  with your topics and create new topics with the output.  And that one can go to your various consumers, for example, databases, reports, dashboards,  applications, machine learning, or storing it in Hadoop, for example.  And you can exchange schema versions through, for example, Git repository or some other  central place where everyone can retrieve these definitions and the developers and engineers  that write the message, they write the input data, they provide their protobuf definitions,  the engineers writing the streaming processors, they also provide the definitions for their  output while reading the one from the engineers and the consumers, they just need to read  the schema definitions to be able to parse everything correctly.  So how would that look like in an example?  Here we have three different systems that emit events to Kafka.  So for example, a web service that writes log events to your stream.  So log events from your website, for example.  In there we have a timestamp field, a session ID and a type, like, okay, use a looked at  a product or a customer put a product in the shopping cart.  And then we have the optional fields, customer ID and product ID.  Then from your customer database, you get these customer messages, which again have  a customer ID, a name and an email.  And your product database provides the product messages.  And with a product ID, a name and a price.  These then you have a stream application joining these three topics and emitting a new message,  which is a very simple message.  My rich event message basically has three fields, one event field, one customer field  and one product field.  So I actually don't need to write a lot of schema definitions because I can just import  the previous ones there.  And the output of this streaming application is again a stream in my Kafka cluster of these  rich events.  And then in that I can, for example, plug in another application that does something  like sessionization, so it takes the session ID out of there and groups by them.  And my session message is also again very simple.  Session message just consists of repeated rich event messages.  So makes it very easy to define these things.  Of course, at this step, I can also do further aggregations like include fields for counting  how many different products a user has looked at or how many interactions they did with  our website and so on.  But it's basically a very simple version.  And that one is then something that you can, for example, read by machine learning algorithms  for personalization or recommendation engines.  You can put it in reports or you can let your analysts read it and do analysts and do their  talk reports on it.  Now going back to this chart, so the skid repository or whatever you're using, I mean,  it's basically just like the Avaro schema registry that comes with a confluent platform  of Kafka.  So why is this one, what's the advantage of this one?  Now it can always happen that not all the protobuf definitions there are actually up  to date, for example, the developer could have forgotten to commit their message or the host  of your git repository accidentally deleted their production database and is now trying  to recover a backup.  So it could always happen that this one doesn't quite work and someone doesn't have all the  schema versions that they need.  So what happens in that case?  Let's, for example, assume that in your product database, someone adds a new field for the  color of your product and starts writing that to the Kafka product stream.  Now your streaming application is starting to, or is processing these events and putting  them in the rich event message.  And all that happens is now that the product contains an unknown field, which is the color,  but everything else will just keep on working.  And even further down, it causes no trouble at all.  My session will still contain these rich events and these rich events all will contain  a product with unknown fields.  And even at the end, the applications can still read these messages.  And if the BI analyst or data scientist at the end looks at the messages, he can find  that unknown field.  He can look at the value and might even be able to find, to determine that it's supposed  to be color.  Maybe he was even the one that wrote the task to the engineer in the first place to put  the field in there.  So the whole pipeline basically still works the same way.  Similar if you just delete a field like here, the price.  For a protobuf, there's no difference between a field that simply doesn't exist and an empty  field with a null value.  They both are simply not included in the protobuf message.  So all you're going to have now is a product that contains a null price.  And since your price since the very beginning was marked as an optional, hopefully the engineers  writing the stream applications kind of put in logic to handle null prices already.  So your streaming applications will continue running uninterrupted.  So in general, what does this allow you?  So it means that the different teams that are responsible for data, for producing the  data, for processing the data, for consuming the data, they can all move at their own speed.  There's no real strict alignment for releases necessary.  So if you have a traditional OLTP system that feeds into a data warehouse and you want to  do schema changes across the system with several, for example, MySQL servers involved, it's  very, very painful because you need to make sure that each of the queries takes that into  account.  But here we can really do a hands-off in data engineering.  The people in the middle writing the streaming process, they don't need to do anything.  They just read the data and write it to the output.  And everyone can basically upgrade to new schema versions at their own speed.  And only the actual producers and consumers need to align on including new information  in your messages and the pipeline in between just forwards them without needing to know  about it.  Of course, not everything is great about Protobuf.  There are also some downsides compared to Avro, so there's no...  Why do you kind of need to hand-craft your schema and be very considerate about it so  you cannot reuse your field IDs?  That's the important thing.  And you really need to stick to them.  And you need to also think about what you may mark as required and optional because  once something is required, you basically can never change it.  There's also less implementations for it around Kafka and Hadoop ecosystem and less people  using it.  So you will find out that you will often have to write the serializer, the serializer yourself.  Therefore, Avro, you basically find enough implementations of that.  A big issue is that Google kind of wants to remove these unknown fields in Protobuf  free.  It's a very long discussion now going on in the related issue on GitHub.  And one more thing is that Protobuf messages usually are one or two bytes larger than Avro  messages.  Nowadays, I don't think that's a really big deal, but for some people that is an important  factor.  All right, thank you.  I hope you learned something about Kafka streams and Protobuf and maybe have some inspiration  yourself.  Do you have any comments or questions?  Yes?  I have a question about navigation.  I have two servers running after that.  And they get each other.  I don't need to send them out or make them.  Are there any way to make sure that all messages that we give to those servers?  Yeah, I mean, if you have a...  So the question was, if I only have two servers in my Kafka cluster and one of them goes down,  how can I make sure that I don't lose any data and things still keep on running?  Well, if you set your topics to all have a replication factor of two, it means each  of your partitions, everything should be existing on both servers.  So in that case, basically, they would be identical.  They both would have the full data set.  And if one of them goes down, the other one still has everything.  And the producers would then all go to that one running Kafka server and write everything  there.  And the consumers would also start only reading from that one.  Did that answer the question?  Yeah.  One producer, I'll just say message.  Yeah.  Either any delay between the message reach the secondary side or...?  I mean, usually the producer sends out the Kafka cluster, sends out an ACK signal back  to the producer that it received the message and processed it.  And you can also define when it does that.  So you can say only acknowledge a message if it has been replicated on both machines,  so you can make sure that nothing is going missing.  That's actually a talk by Gwin Shapira about when messages absolutely have to be there.  If you want to have a very strong guarantee that your messages are there, recommend looking  that up.  It's very helpful in that topic.  Yeah.  Yeah.  Yeah.  I have a question, extension of this question actually.  So if you have Kafka clusters, I mean, two data centers.  Yeah.  I'm actually after you.  With that work, you mean these two data centers and the other one, you mean the one that you  have to be able to do?  Yeah.  Yeah.  Yeah.  Yeah.  Yeah.  Yeah.  Yeah.
Kafka_Streams_and_Protobuf_stream_processing_at_trivago_0_131.png Kafka_Streams_and_Protobuf_stream_processing_at_trivago_132_491.png Kafka_Streams_and_Protobuf_stream_processing_at_trivago_492_2399.png Kafka_Streams_and_Protobuf_stream_processing_at_trivago_2400_3215.png Kafka_Streams_and_Protobuf_stream_processing_at_trivago_3216_6923.png Kafka_Streams_and_Protobuf_stream_processing_at_trivago_6924_8807.png Kafka_Streams_and_Protobuf_stream_processing_at_trivago_8808_8867.png Kafka_Streams_and_Protobuf_stream_processing_at_trivago_8868_11975.png Kafka_Streams_and_Protobuf_stream_processing_at_trivago_11976_13631.png Kafka_Streams_and_Protobuf_stream_processing_at_trivago_13632_13715.png Kafka_Streams_and_Protobuf_stream_processing_at_trivago_13716_15455.png Kafka_Streams_and_Protobuf_stream_processing_at_trivago_15456_15503.png Kafka_Streams_and_Protobuf_stream_processing_at_trivago_15504_15551.png Kafka_Streams_and_Protobuf_stream_processing_at_trivago_15552_15599.png Kafka_Streams_and_Protobuf_stream_processing_at_trivago_15600_15647.png Kafka_Streams_and_Protobuf_stream_processing_at_trivago_15648_15695.png Kafka_Streams_and_Protobuf_stream_processing_at_trivago_15696_15743.png Kafka_Streams_and_Protobuf_stream_processing_at_trivago_15744_15791.png Kafka_Streams_and_Protobuf_stream_processing_at_trivago_15792_18071.png Kafka_Streams_and_Protobuf_stream_processing_at_trivago_18072_19319.png Kafka_Streams_and_Protobuf_stream_processing_at_trivago_19320_19859.png Kafka_Streams_and_Protobuf_stream_processing_at_trivago_19860_20171.png Kafka_Streams_and_Protobuf_stream_processing_at_trivago_20172_21479.png Kafka_Streams_and_Protobuf_stream_processing_at_trivago_21480_21527.png Kafka_Streams_and_Protobuf_stream_processing_at_trivago_21528_23735.png Kafka_Streams_and_Protobuf_stream_processing_at_trivago_23736_25307.png Kafka_Streams_and_Protobuf_stream_processing_at_trivago_25308_26411.png Kafka_Streams_and_Protobuf_stream_processing_at_trivago_26412_27239.png Kafka_Streams_and_Protobuf_stream_processing_at_trivago_27240_27323.png Kafka_Streams_and_Protobuf_stream_processing_at_trivago_27324_28523.png Kafka_Streams_and_Protobuf_stream_processing_at_trivago_28524_29603.png Kafka_Streams_and_Protobuf_stream_processing_at_trivago_29604_31451.png Kafka_Streams_and_Protobuf_stream_processing_at_trivago_31452_33527.png Kafka_Streams_and_Protobuf_stream_processing_at_trivago_33528_37043.png Kafka_Streams_and_Protobuf_stream_processing_at_trivago_37044_37127.png Kafka_Streams_and_Protobuf_stream_processing_at_trivago_37128_37158.png