It's like 10,000 streams when what you need is a queue

00:00:00 Hazel Bohon

Hello again, everyone, and thanks to everyone for joining us for another Particular live webinar. My name is Hazel Bohon. Today I'm joined by Chris Simon. Chris was co-founding CTO of two startups in Australia, where he oversaw the architecture and development of complex event-driven systems for over a decade.

More recently, he provides domain-driven design training and consulting, and founded Contextive to support teams with complex domain languages. Chris has a great talk for us today. It's Like 10,000 Streams When What You Need Is a Queue. For all of the attendees, please use the Q&A for any questions.

We'll collect them through the presentation and then answer them in a Q&A at the end of the presentation. Today's webinar is recorded and we'll be sending a link to all attendees after the event. Okay, let's talk about streams, queues, and event stores. Chris, welcome to Particular webinar.

00:01:05 Chris Simon

Excellent. Well, thank you so much, Hazel, for the introduction and to Particular Software for having me to you today. I was going to say this morning, 'cause for me, I'm in Australia and it's currently 2:00 AM, but I think we have people from all around the world, so whatever the time of day is for you, thank you for joining.

As Hazel said, this talk is titled It's Like 10,000 Streams When What You Need Is a Queue. And I'm curious, it's not going to be possible for me to see the answers to this, but I always wonder when I deliver this talk, how many people actually catch the reference here? I suspect there's probably some people here from North America, so maybe it's a higher proportion than I've had in the past in some European venues. But of course, this is a reference to a wonderful song.

If you scan this QR code, you can queue it up on YouTube and have a bit of a listen to it later. It's that absolute '90s banger, Alanis Morissette with Ironic. Now, maybe you're not as gray-haired as I am, but if you remember this, Alanis is in a car. There's four versions of her with four different colored tops. She's driving across the Canadian landscape with the snow and singing about all of the things that she's experienced in her life that are deeply ironic to her.

Again, if you remember the '90s, you might recall that there was some deep debate back then over whether this song was actually ironic or merely just a list of things that were very annoying. I used to be in the camp that it was just annoying and that Alanis didn't understand irony. I've since changed my view. I'm now Team Alanis. I think that actually she understands irony better than any of us. And so that's why I thought it was a nice tribute to reference this song with this talk.

Give you an example of some of the things that she sings about. She says, "It's a black fly in your Chardonnay, or it's a traffic jam when you're already late." Or, "It's like 10,000 spoons when all you need is a knife," which of course I have repurposed the saying, It's Like 10,000 Streams When What You Need Is a Queue.

Now, I'm not going to subject you to my singing voice. That would not be something that anybody would enjoy. But we are going to reference these lyrics a couple of times through this talk, just to point out some of the deep ironies that I see in the way that people approach event-driven architecture.

So why are we talking about streams and queue at all? Well, of course it's about this term, event-driven architecture. If you've not used event-driven architecture before, it's an approach to systems design, to your system architecture that is centered around the concept of an event. We often call it a domain event 'cause it's supposed to represent that something that actually happens in your business domain. It might be something that is expressed as a noun verb. So usually we say the thing that something has happened to and the thing that has happened to it.

We try and use this past tense because we're trying to make it easy for us to work out the sequence in which things happen. So we might say something like, "Order submitted," or, "User registered." These are events that capture a moment in time, where at one point, the user is not yet registered and then the user has become registered. So this moment where they transition from not being registered to registered is what we are representing with the concept of this event.

We often find when we're thinking about events, that we want to think about what causes those events to occur, and we call those things commands. So "submit order" is an example of a command. It's not expressed in the past tense. It's expressed as an imperative, an instruction to the system. "I want to submit this order now." I mean, usually coming from the user, but not always, sometimes from different systems to each other.

One of the best ways to learn about the events in your business domain is a practice called event storming. I don't know how many of you have actually used event storming. It was invented by Alberto Brandolini, he's from Italy, I think 2013. It's a fantastic technique. You gather many people from across your organization, people from the engineering team, from the product team, from customer service, from sales, even from perhaps legal or other departments that work with your customers or are involved in running the business in some way.

People from the shipping department, from the warehouse. Anybody that's involved and actually understands deeply the way that your business operates. You all get together in a room and there's a facilitated process that guides this rabble of people that may or may not know each other very well into capturing the time series of these events on a wall. It represents the generalized process flow of your business. It's not intended to capture anything particularly precise.

It's not about precision. Alberto used to work in teaching people about UML, and realized that UML presented so many obstacles to collecting information and knowledge from people that hadn't been trained in it. So he tried to find a technique that was very simple and easy for people to understand. You could enter the room knowing nothing about how this process was going to work and still contribute immensely valuable knowledge from your corner of the business.

It's a fascinating technique. I recommend you look it up if you're not familiar with it, but I don't have time to go into too much more detail on that one today. The end result though is this nice time series. We start at the left, we have a command to register a user. And if it all goes well, the user will have become registered, so we can have an event that represents that. It might be adding items to cart or removing them from carts.

And so we end up with events, item added to cart, item removed from a cart, those past tense representations saying that, "This is now a fact about the world. This has happened" as we move from left to right. We submit the order and the order has become submitted. We might then have a series of events that are not linked to commands because they occur as a natural consequence of the events to the left of them occurring.

So the payment is successful or failed, then the order becomes accepted, shipped, and eventually received by the customer. This is, of course, a dramatic simplification of a real business process, but it helps give you a visualization of how this kind of thing can occur.

Once you have this list of events, it's actually a very helpful way to start to think about how to decompose your system into what domain-driven design calls bounded contexts. We draw these boundaries around different parts of the workflow to try and encapsulate the things that relate together. Things that are part of a consistent and coherent language and conceptual model so that we can divide up the system and allow different teams to work on and own different parts of the system.

So these bounded concepts are an incredibly important concept from domain-driven design. And event storming is just one of the ways that you can start to get a handle on, to think about, to make decisions about how you're going to decompose your system into these bounded contexts. Again, it's a very simplified model, but for this example, we've ended up with a shopping cart context, a payments context, and a shipping context. Quite a standard representation of a typical e-commerce platform.

Now that we have bounded contexts, we can start to think a little bit about these events as not just representing a fact or a moment in time, but about an information flow between different bounded contexts. And we start to think about these bounded contexts as playing different roles in that information flow. When we talk about events, we talk about publishers and subscribers. The publisher is where the event originates. It's the owner of the event. It's the thing that has the authority and decision-making capability around whether or not this event has actually occurred.

But then we might have other bounded contexts that subscribe to that event. They would like to find out when the event occurs. So the event can become a flow of information, a notification between bounded contexts. And with a command, it's the inverse. We have a handler of that command, the bounded context that's responsible for fulfilling the intention of the command. And then other bounded contexts can send the command in. They become senders of that command. This language that we're using here is quite important and quite precise. We have publishers and subscribers when we're talking about events and senders and handlers when we're talking about commands.

Now, this highlights an interesting point about event-driven architecture. Many people think that it's about designing more loosely coupled systems, and it certainly can help you with that. But using events doesn't implicitly eliminate coupling. It just reverses the direction of it. So, if we think about a command, then it's quite analogous to using an API. We're having multiple clients of the API being able to invoke an API, which is kind of logically equivalent to sending a command into the handler.

And so in this case, the handler is not coupled to its senders, but the senders have become coupled to the handler. They've taken a dependency on it. When we publish an event, we invert that relationship. The owner of the event, the publisher is not coupled to any of its subscribers, and the subscribers become coupled to the publisher. So there's still coupling, it's just about the directionality of it.

Now that we have these events, eventually we're going to have to implement a system. We're going to need operational systems to capture those events, to store them, to manage the flow of that information between different parts of the system. And there are a couple of very common technologies in play out there. So we have to make a choice. We might find that we are using something like a Kafka system and assigning events to topics in a Kafka system.

Or perhaps we're using a queue-based pub/sub system and assigning events to, depending on the system that you use, maybe they call exchanges or even topics in those systems as well. Or perhaps, you're thinking about it differently and you're using event sourcing. And you're thinking about your event store and events being records that are associated with entities in your system, streams associated with those entities, the stream of events that affect those individual entities being captured into an event store.

I can't see your faces, so I'm not sure which one of these is resonating most strongly with you, but I typically find that there's a diverse mix of people that default to assuming that this event, this orange sticky note, which is represented on this Miro board, but in real life is an actual orange sticky note, is going to end up being a Kafka topic, some sort of concept in a queue-based pub/sub system, or perhaps a record in an event store.

Maybe it could be any one of them, maybe it could be all three. That's what this talk is really all about. We're getting to the heart of how we make this decision, how we decide for each of our orange sticky notes, each of our domain events, how we're going to represent it in our runtime operational system.

A really key concept that I use to help me think about this is this thing called Christian's 4+1 views. Philip Christian was a systems designer working in the '90s, and he had recognized that talking about systems design was very difficult because many people in your organization have completely different perspectives. And so he decided it was important to actually capture those different perspectives explicitly.

And he came up with these different views. These are all different ways of describing the same system. At the top here, we have the Logical View, which relates to the domain. It's what domain-driven design is all about. It's about bounded context, entities, commands and events, and just concepts that come from the business domain without reference to the specific technology implementation.

Then we have the Process View, which represents the runtime artifacts in our system. Maybe they're the microservices or the APIs or the databases, the things that are actually executing in a live running system. We have the Physical View at the bottom, which is the infrastructure that those processes are executing on, and the Development View, which is the way that we look at the system as a collection of packages, and maybe in .NET, assemblies that are depending on each other is sort of like call-chain graph of concepts.

And maybe we're thinking about it as the folder structure or the package dependency list. Either way, we're thinking about it from the perspective of how a developer is looking at the system as they're working on it. And these are all four very valid perspectives of the same system, and it's important to consider all of them. I typically find in my work that almost everyone spends a lot of time thinking about the Process View and not nearly enough time thinking about the Logical View or even modeling explicitly the Development View. So this is an interesting idea and I recommend checking it out more if you're curious about that.

The reason that we bring it up today is because we've talked a bit about orders and events. Sorry, we've talked about events, ordering submitted being an example of it, which is something that lives in the Logical View. Everything we're doing in our event storing process is about capturing the Logical View. But this decision about the Kafka topic or queue-based pub/sub or event store is something that's about the Process View.

And this distinction between the two, and keeping clear in your mind that they're different decisions, and you're doing different modeling activities and different design activities when you're thinking about these different things is, I think, quite helpful. But to make that decision, how are we mapping from this logical event over to the runtime Process View? We need to understand the differences between these technologies, the difference between a Kafka topic and a queue-based pub/sub system. And then later in this talk we're going to talk about event stores because it is technically a different category of thing.

Let's get into it. What are the differences really between queues and streams? When I talk about queue-based systems, I'm talking about examples like RabbitMQ, Azure Service Bus, AWS, SNS and SQS. And when I was talking about streams, we're talking about Kafka, Azure Event Hubs or AWS Kinesis. I've mentioned Google Pub/Sub as being on both sides-ish. We're not going to dig into the nitty-gritty of that today, but it does have characteristics of both.

I'm going to start by looking at what do they have in common? And the first thing is that you can think of them as message brokers. It's a generic term for anything that stores and facilitates message distribution, something that allows us to get messages between different runtime processes in our system. Another thing that's common is that they all have topics. I referenced this before. We can organize those messages according to topics. Some systems call them channels or exchanges. And associated with the topic, we have some ability to store messages for some period of time, usually not forever. These are not intended to be databases. They're intended to store the message for as long as it's needed from when the publisher produces the message to when the subscriber picks it up.

We have producers and consumers. Producers are the components that push messages into the broker and consumers are the components that consume messages from the broker. And again, thinking about that distinction between Logical and Process View, I made a big point of talking about senders versus handlers and publishers versus subscribers before. That's language of the Logical View. The language of producer and consumer belongs in the Process View because it's a bit more abstract. It doesn't refer to the intention of the messages.

It's just about the technical mechanism of producing and consuming messages. So when I hear people talking about message producers and consumers, I know that they're focused on the runtime characteristics. They're not so much focused on the logical meaning and intention and business domain characteristics of these events and messages. But this kind of language, I think, and the precise use of it can be very helpful to make sure you're communicating clearly about the kind of design decisions you're making in your system.

Another thing that they have in common is the concept of delivery guarantees. We have producers who can produce messages and consumers consuming them, and we need to think about what happens if messages are dropped, in which case we have At Most Once delivery guarantees. Message two in this case is dropped and the consumer can only receive message one. We have Exactly Once, where for every message that's produced, it's processed by the consumer exactly one time, and At Least Once where a message may get consumed more than once.

This is actually a very deep topic. You could do a whole one-hour talk just on this and probably much more, particularly around Exactly Once processing because there's a lot of nuance to it. And I don't want to dismiss the complexity here, but again, that's not the purpose of today's talk. But if you're interested in things like the outbox pattern and so on, they can help you with this. That's what they have in common. Rushed through that quite quickly because those concepts are relatively simple. And I want to take most of my time today talking about what is different.

We're going to talk about what's different from four different perspectives. The way in which these systems distribute the messages, the way in which you handle the scale out of consumers when you are scaling out and having multiple consumers trying to process messages in parallel. The way it handles message processing failure. If there's an issue process handling a particular message. And the concept of message routing, which is subtly different from message distribution, in that it starts to relate to decisions being made around who should receive specific messages. Let's start with message distribution.

What I mean by message distribution is who decides if the payments context should receive this message that is being published by the shopping cart context? There's a decision that needs to be made over whether this message should become visible, and queues and streams manage the state of this message very differently. When we're talking about queues, the broker keeps track of it for you. It characters the state in itself. And when we talk about streams, the state is managed by the consumer. The broker does not manage that state for you. It might be a little bit abstract, so let's dig into it with a particular example.

We have to look down into the details of what is the interaction between the consumer and this broker system. So when we're using a queue, the interaction has the semantics of pop off the queue. It's a stateless interaction. It doesn't carry any data in the request. It just issues a request to the queue and the queue returns back the next message that's available, this sort of pop operation. But when we're using a stream, which we more properly should be calling a Distributed Log, what we're actually saying is read from offset X.

And this offset X tells us or tells the stream which messages to return, any messages that have appeared after offset X. And it's this offset X that I'm talking about when I say that it's a stateful operation. So, while the production, the producing operation is relatively similar, we're pushing to queues and appending to logs. From the producer side, it looks quite similar. On the consumer side, we're either popping off the queue, a stateless operation, or we're reading from an offset, which is a stateful operation. And this affects many aspects of how these two systems work differently.

We can look at a more concrete example here. So let's say we have a queuing system and the streaming system, our shopping cart is pushing or pending message one into these two different systems. And then it pushes message two in. Then in the queuing system, the consumer will pop off the queue and receive message one, which leaves only message two remaining in the queue. Whereas with the streaming system, the Distributed Log, we read from offset zero and we receive back message one.

And then we update our state to say, "Offset one is the most recent offset that we've successfully read." Maybe then a message three is pushed in on both systems, and then we do another read operation. In the queuing system, we pop off the queue and we get back message two, leaving only message three in the queue. Whereas with the stream, we read from offset one, which was what we stored in our state, and get back offset two.

We update our state to set offset equal to two. So that's the most recent message that we read. And then we can read the next one. Pop off from queue, we get message three, read from offset two, and we get message three. So in both systems, we've had the successful transfer of three messages from the producer to the consumer, but the way in which it happened is significantly different because of the state management.

And at the end of the operation, the queue is now empty. And that's what I mean by the state. The queuing system is managing the state for you. It's now empty because it knows that it should not longer return any messages to any consumers that attempt to read them. But the Distributed Log, the stream, has kept all of the entries. It's not tracking the read state of each message. It's just storing the messages and relying on the consumer to request the messages that it wants.

So this is a really fundamental difference, and you really need to internalize this to be able to articulate and decide between these two different types of systems. And so I refer to them as stateless consumers and stateful consumers because that state is a key part of the consumer. If you've used Kafka, you might be a bit confused at this point and thinking, "Hang on, Chris, I used Kafka. I didn't have to worry about state management at all. I didn't write any code to care about keeping track of this state. I did not have to do this thing that you're saying I have to do."

And this is because the client libraries that have been released to work with Kafka are doing it for you. They're actually storing that state back into a topic in Kafka itself, but it's managed by the client, what process, even though it's in the library. So it's not in your code, but it is in the process. And because it's in the process, that's what affects other things like failures and scale-outs, as we're about to see.

And the fact that it stores it back into Kafka does not mean that the Kafka system actually really knows about that data or does anything meaningful with it. It just provides a storage mechanism for the client library. It would be possible for the client library to store those offsets into a completely different data store. But of course, for operational simplicity, it makes sense for it to just write it back into Kafka because it does have Kafka sitting there and it can use it for this purpose.

Which leads us to one of the first ironies of these sorts of systems in event-driven architecture. When you look at the marketing literature out there around Kafka, it makes a big point of referring to this kind of system as data in motion. And it talks about the importance of thinking about, "Oh, our data's not static anymore. Our data is in motion."

But when I think about the way that I visualize those two systems, I think about queuing systems, where messages are flowing into the queue and then out of the queue, and leaving the queue, versus systems where the messages arrive into the streaming system and stay there, and just live there until they're read by the consumer. But then even after they're read, they're still there sitting there in the stream. I find myself thinking that, actually, data in motion applies far more effectively to a queuing system than it does to a streaming system.

And so returning back to Alanis, who's going to help us understand the vagaries of these things. Isn't it ironic? It's like your data's staying still when you though it was in motion. Okay, I never promised to be a lyricist or a singer, but I am trying to reference these things. And this is Alanis in her yellow jumper now. You can't see too much snow in this picture, but it is there 'cause it's Canada, of course.

All right, moving on to consumer scale-out. When we talk about scale-out, we're talking about multiple copies of the same logical consumer. So as visualized here, we have a shopping cart producer and we have a payments consumer, but we have three copies of the payments consumer. And because those consumers are stateless with queues and stateful with streams, there's an impact of that state management on how we think about scaling out these consumers.

When we think about stateless versus stateful clusters of scaled-out nodes, now there's actually quite a parallel here to traditional systems using web requests and load balances. If you think about a load balancer that can route a web request to any node, that's a stateless system. And in the world of the cloud, this is kind of every system. We've really adopted elastic scaling. We've got load balances that can dynamically route requests to any of the nodes that are currently in the cluster.

The state still has to go somewhere in these systems. It's just we've externalized it into data stores so that our web tier can scale out elastically and we can add and remove nodes from those clusters seamlessly. If you remember the bad old days, we used to use things like content-based routing, which was some kind of a session ID cookie or some other payload or web hash because we would often have in memory storage.

We used to have session states stored in memory quite commonly in systems, which really restricted your ability to scale out your web tier, because once there were states stored in each node of the web tier, each of those nodes became special. And it was up to the load balancer to make sure that the data would flow through to the correct node, for example, using something like a session ID cookie.

I said, if you remember the bad old days. Someone reminded me the other day that this is actually the bad current days for many people working with legacy systems, so I do apologize to you. But obviously, we've made conscious choices over the years to move away from this kind of stateful web tier processing into stateless processing 'cause of the flexibility it offers us with handling our web tier node clusters.

But the truth is that because of the differences in how we think about state and the way that that affects our scaling capabilities, stateless systems are more like queues and stateful systems are more like streams. And when we think about stateful clusters, there are many examples that we can turn to for other stateful systems like databases, techniques like sharding that allow us to allocate specific pieces of data to specific nodes in the cluster. Or traffic type splitting, where we have read and write replicas in a database, or different kind of workload allocation, where we have different roles for different nodes in the cluster, leaders, workers, and so on.

And streams use these sharding patterns, or they call it partitioning, and a form of workload allocation. It's not exactly as it's illustrated in this diagram because the nodes don't actually talk to each other, but there is role allocation in a cluster of stream consumers. So let's again look at our example of messages flowing through the system and see how this plays out. Let's imagine we have one message already in both of our brokers.

In a queuing system, both of the cluster nodes will issue a pop-off queue request, but only one of them will receive the message. Because the state is being managed by the queue, it will make sure that only cluster node one will receive that message one. In a streaming system, if we use naive consumers that are not using many of the capabilities offered by the client library, sometimes these are called low-level consumers, and they're sharing an offset management story, then they will both read that message, because they both will request read from offset zero and they'll both get back message one.

So we end up with the message being processed twice, or at least as many times as we have nodes in the cluster. And again, this is if we're using the Kafka low-level consumer because the high-level consumer has embedded some smarts to protect you from this scenario, because typically multiple processings of the same message by parallel nodes in the same cluster is not typically what you want. If you imagine that this is a payment service that is reading this message and initiating a credit card transaction, you would not want it to occur twice or three times if you had three nodes in the cluster.

In the queuing world, we call this concept the competing consumer pattern because these consumers are competing with each other to get the next message off the queue. But in the streaming world, we don't have that competing consumer concept. It's up to each node to include a state data, the offset that it wants to read from in each request. So we have to find a smarter way of doing this.

And much like in stateful databases where we use sharding in stateful streaming systems, we use partitioning. So in this case now, the producer has to make a decision. "We're going to partition our Kafka system into multiple partitions, and we have to decide which partition are we going to add each message to." So there's a partition selection algorithm on the producer side making a decision, and it's important that you want to keep this balanced. You don't want to have a hot partition where all the messages are ending up on one and not the other. So you have to understand the distribution of your messages and how this algorithm is going to be affected by that.

In the queuing system, while a lot of queuing systems do support similar concepts of scaling out and partitioning, it's typically transparent to the consumer. You push the message in and you don't worry about it, you let the system take care of it. And with our Kafka high-level consumer, it actually is able to be partition-aware. And so we can decide node one is allocated to partition one and node two is allocated to partition two, and we're going to keep track of the offset for each partition independently.

So perhaps now that we've read message one, node one will have offset one stored and node two will have offset zero stored. Let's say we add a third message into the system and let's start looking at the consumer operations. So in the queue, we have pop off the queue coming from node one and two, and each of them receives a different message. Node two will receive message two. Node one will receive message three in the streaming system.

Even though they are both saying, "Read from offset one," because the messages are split between the partitions and the nodes are reading from different partitions, node two will read, "Offset message two," and node one will read, "Message three." And now we see we can actually have each message only processed once because we've split the messages into partitions and we've allocated each node to our partition. And again, the Kafka high-level consumer client library is handling this partition allocation for you automatically.

If we push a few more messages into our system, then we can start to think about what happens if something goes wrong with one of our nodes. In the queuing system when node two fails, well, node one will pop off the queue and read message four. But in the streaming system, node one will say, "Read from offset three" and it will get message five. It skips message four because message four is on the other partition, which it's not looking at, and node two is currently offline.

And so if you don't have something in here to manage this, message four will never get processed. And again, that's with the low-level consumer because the high-level consumer client libraries will protect you. There's actually a process where the partitions are rebalanced. There's a leader that will instruct node one to be reallocated and to start reading from both partitions one and two once it detects that node two has stopped reading, and eventually message four will get read by node one.

So, there are mechanisms to protect you, and in the end we can have the same result, that all the messages eventually are processed, but the mechanism is different, and it's important to understand that because it does play out into different characteristics. So, one of the most important ones is what happens when we scale out further? Well, in the queuing world, if we add a third node into our cluster, it immediately adds to the processing capacity of the cluster.

It will start competing with the other two and the queuing system will just start returning messages to evenly distribute them across messages nodes one, two, and three. In the streaming world, if we add a third node, well, probably that will be okay, but if we only have two partitions, the third node's got nothing to do. So it's really common in streaming systems to over-allocate partitions based on what you think the scaling limits are that you're going to be dealing with, and hopefully you get that estimation correct.

It's certainly not worth underestimating this. In fact, modern systems can handle many more partitions than you're likely to need, and so that is a common pattern. You allocate tens or potentially 100 partitions and allow the nodes to be reading from multiple partitions at once. And then as you scale up and add more nodes, you rebalance those partitions so that each node can get access to some subset of them.

And eventually, you will hit a limit. If you scale up and scale up and scale up, you'll hit the point where each node is allocated to just one partition, at which point you can scale up no further. You cannot add any more nodes effectively until you reallocate partitions within the stream. You have to add partitions to topics. And at the end, there are systems and mechanisms within Kafka to help you do that, but it's something you need to be aware of and something you will need to manage.

So Alanis, can you help us out with another irony? Isn't it ironic? It's when you need to scale out, but your partition counts too low. That would be a shame if you were in that situation. Let's talk about individual message processing failure. We talked about node failure in the last one because it affected how we think about the scale out of consumers in that cluster. But now, we're going to talk about individual messages.

If we have our producers producing messages, so message one into the queue, message one into the Distributed Log, message two. I did gloss over one detail before. With the queuing system, when we pop off the queue, we don't immediately delete it from the queue. While the message is being processed, the queue manages the state of that message for you using a invisibility state. It will say that, "I'm going to keep the message, but I'm not going to release it to any other nodes or any other requests until I'm told that everything's okay with this message."

The equivalent in the streaming world is that the client, the consumer, can control when it updates that offset storage. So in the queuing world, we have to wait until we've successfully processed the message, and then we acknowledge it back to the queue, which effectively results in it deleting that message from the queue as a confirmation that it's been successfully processed. And in the streaming world, to confirm that it's been successfully processed, we update that state storage to say now, "Okay, I've done message one." So we start reading message two. It goes invisible in the queue. It's being processed in the stream. We haven't yet updated the offset.

And what happens if message two fails? So maybe message two has a schema issue, we can't de-serialize it, or perhaps it requires that in the processing of that message, there's a necessity to interact with a downstream dependency, a database or a third-party API that's temporarily offline. Well, we're sort of stuck here. We're not processing that message. There's some sort of failure. And maybe it crashes the node or maybe the exception gets caught and handled. It depends on the specific system that you're using.

But either way, in the queuing world, because it's keeping track of the visibility of that message, it will eventually say, "Hang on, this message has been invisible for quite some time. Possibly something's gone wrong in the consumer. I'm going to make it visible again." The specifics of how this work vary between different systems, but the concept is the same. There's some sort of timeout or trigger to say, "This message has not been acknowledged. I'm going to surface it again because I'm going to assume that there was a failure." The queue's taking care of that kind of decision-making for you.

Whereas in the Distributed Log, there's actually a situation where you can end up by default skipping that message. There's many different places this offset is stored. For a little while, it's kept in an in-memory cache, and then eventually it's flushed out to the offset storage. So depending on when the failure occurs, whether the exception is caught, if it causes the node to crash, all these different things can lead to different outcomes. Actually, node crashing is the best case scenario, because when the next instance of the node spins up or the replacement node spins up, it will start reading from the offset and it will replay the message.

But if the node doesn't crash and you don't write your code and you handle the code correctly, there can be a risk that you can end up, just by default, skipping that faulty message because of that offset having been incremented in memory even before the successful storage has been updated. Whereas with the streaming system, we get a second chance to process message two. And there's a real focus in the design of the queuing systems and brokers to make sure that each message is really important. It doesn't want to let it go until it's confident that it's been successfully processed.

Again, because in the streaming world, the Distributed Log is let go of that responsibility. It's really up to you to make sure that you're using the client libraries correctly and configuring them correctly to get the scenario that you want. Maybe in some cases it's okay to skip that message. Maybe you're dealing with a stream of messages where individual messages don't matter so much, and what you're more worried about is the most recent message, in which case that's fine. But if it's not okay, then you have to take care of that. You have to make sure that you are configuring things using the appropriate libraries, and using them correctly, to get each message processed successfully.

And one way of doing that is you're going to explicitly seek back to that failed message. But like I said, there are libraries out there to help you with that situation. Eventually, the message might get processed correctly. Maybe the downstream database comes back online or the downstream API. The consumer can process the message and acknowledge it back to the queue. And in the streaming world, we can finally update that offset and be confident that the message has been processed.

Another interesting scenario occurs, though, in the situation where the same message just fails over and over again. We often call this a poison message because it's never going to be successfully processed. The reason it's failing is not because of a downstream dependency. It's probably 'cause there's some fault in the system itself, maybe a schema, a change that's leading to a de-serialization issue, or a bug in the system that's activated by the specific data that's in that message.

In a queuing world, after repeated failures, the broker tracks that. It'll take track of how many times it's attempted to process that message and eventually forward it to a dead letter queue. It treats it as a poison message. It gets it out of the flow, stops releasing it, and it gives you the opportunity to monitor that dead letter queue and to action any messages that arrive into it, and to investigate them, to understand why they ended up there, what caused it, and to take appropriate action to make sure that eventually that message does get processed.

In the streaming world, once again, maybe you're skipping messages or maybe you're retrying them, but if you are retrying them, then you need to keep track of how many times this message has been processed because the streaming service is not doing it for you. And you need to make sure that you take care to configure dead letters, topics, and have the appropriate logic, again, using the right libraries if they're available in your ecosystem about what to make that decision about, what to do when you've had too many retries that have failed, and how to handle that eventually.

So, you can see that it starts to feel like handling failures and handling repeated failures, in the streaming world, we end up with these things that kind of feel a bit like workarounds. They're not baked into the core system. It wasn't a key design criteria or a design goal of the streaming system to have this situation where we really care about every message getting processed successfully at the end of the process.

And if you don't do things correctly in a streaming world, it's very easy to end up with this phenomenon called head-of-line blocking, where one poison message can actually just sit there blocking the rest of that partition, because every node that spins up to try and read the next node off the partition causes a failure and a termination of the process, a crash. Another node spins up, reads the same message, crashes. Another node spins up, reads the same message and crashes. And if you are not configured to deal with that, you can block the partition. It's this thing called head-of-line blocking.

So Alanis is going to help us out again with another irony of event-driven systems. "Your dev team adopted EDA, got head-of-line blocking on their very first day. Isn't it ironic, do you think?" is what she would say to us if she cared about event-driven architecture. Let's now talk about message routing, how we actually decide who are the appropriate consumers for a given message. When we talked about queuing systems before, we focused on the queue, but as I mentioned right at the start, is that there often is also this concept of a topic or an exchange.

Now, in queuing systems, the topic and the queue are separate to each other. With streaming systems, they are one and the same. What do I mean by that? The mechanism that's used to decide the distribution of which messages are supposed to go to which consumers, and the storage of the message, is the same thing. So in streams you have a topic. The topic is the unit of distribution, and it is also the location of storage. In queuing systems, the topic is the unit of distribution, but it does not store the message. It forwards it onto the storage medium, which is what we call the queue.

This separation gives you some flexibility of your topology of how you connect these different nodes up. So, to look at it in the simple case, when we're looking at just a single message, maybe we publish or append message one, it flows into the topic and then into the queue. And in the streaming world, it goes straight into the topic and stays there, as it did for all the other scenarios. We pop it off the queue, we read from offset zero as we've seen, and the message flows through to the consumer.

But now let's look at what happens when we have a second logical consumer. So not just another node of the payments context, but a different type of logical node altogether, maybe a notification service or notifications context. In the queuing worlds, the topology that we use here, it is a quite typical one. You don't have to use this, but it works very well, is you allocate a queue to each of the logical clusters.

We call this the input queue for that logical service or that compounded context, and we allocate the topic to the event type. So the domain event gets its own topic, and there's this fan out concept, where we can explicitly connect the queues to the topic to say, "These services are interested in this event." That's what we're saying when we connect those queues to that topic.

In streams, the broker has no real concern around this. It's up to each of the clients to decide which topics it's going to read from. In these different worlds, again, different language creeps in. In the queuing world, we often call these, as I have been doing, logical subscribers. We might just use the term bounded context and implicitly know that that refers to a scaled-out cluster of nodes. But in the Kafka world, they often call these things consumer groups.

And I find it interesting the different perspective that the language reveals here, because in the queuing world, the logical subscriber is a single thing. And the fact that it is a collection of nodes under the hood is kind of an abstraction. Whereas in Kafka, the consumer is the individual node. And so a collection of consumers that have the same purpose is called a consumer group. So we go from the other direction. We focus on the individual nodes, and then we start talking about groups of consumers to talk about the cluster. Whereas with queues, we focus on the logical subscriber as the thing, and don't have to worry too much about what it's comprised of because the individual nodes take care of themselves through that computing consumer pattern.

Let's look at how this fan out works in the queuing world and what happens in the streaming world. In the queuing world, let's go back to that, we publish to the topic, and because of the connection between the topic and the queues, we end up with two copies of the message, one in each of the input queues. Whereas with the streaming world, we just have one copy of the message in that topic storage.

This distinction is interesting because some people think, "Well, two copies of the message, that's going to add a lot of storage. Right? I don't want to have to store multiple copies of each message." But the truth is it's typically not a big deal, because remember, these messages are not sitting there for very long. They're getting processed and deleted. Whereas if you do want to store the message for a very long time, like what happens in a streaming system typically, it makes more sense that you would care about the storage and you would optimize for just having a single copy of the message.

So you can see how the goals of the systems work different, which led to the design decisions that play out in these ways. When we look at the consumer world, the payment service is popping off the queue, reading from its own queue, and it gets that message one and processes it. And the notification service is popping off the queue and getting message one and processing it. And that's actually what you want. In this case, you do want those messages processed twice because we're talking about two different outcomes occurring. In one case, we might be charging the credit card gateway, and in the other we might be sending an email or an SMS notification.

And in the streaming world, because each of these clusters have different offset storage by default, even though they're both reading from the same partition, because they're tracking the offset independently, they both get back message one and do what they need to do with that message. And so this works quite well in both systems at the end of the day. Each consumer group or logical subscriber is able to receive the message that it needs and process it.

If we go back to this distinction between the Logical View and the Process View, and remember that in the Logical View, we think not just about events, but also about commands, and this idea that we might end up in situations where you have multiple senders wanting to send a command into the handler. In the queuing world, there's actually a very natural topological analogy of this, which is that the commands could just get put directly onto the input queue of the handler.

So if we have a command that's offered by the payment service, we can actually bypass the topic altogether, or the exchange if you're using RabbitMQ. Just bypass it all together and put this message directly onto the input queue. Because remember, the senders, they know about the handler. They're coupled to the handler. They can be aware of the existence of this input queue. They know that they want the payment service to handle this message, and they can just put it directly onto that input queue.

Whereas with streams, it's just another topic. There's no semantic or logical distinction between whether the messages represent events or commands. We're just saying, "It's a topic, which is a storage medium. It's a location where consumers can come and get messages from." So we have to make sure that through semantics or access control lists, we make sure that only the payment service is going to be reading this particular topic.

Because typically, when you're looking at a collection of topics, the default assumption is that any service could come along and read from any topic because they represent events. And that's one of the beauties of event-driven architecture, is that you can add subscribers without having to update your publishers. But for these topics, it's actually quite important that we don't allow new subscribers to start reading from them, because logically we actually only want that message to be processed by the one service. In this case, the payments context.

Now, what this leads to is a very interesting phenomenon, which is because once you start to think about topics as storing events, and then eventually you decide, "Well, actually we also need a command to flow through the system," you end up with quite bizarre framing in some of the literature around streaming systems, which you can see here it says, "Each workflow step is initiated by a command event." I just want to pause for a moment there and get everyone to just really think about what does that mean, a command event.

Considering how starkly different these two concepts are, the idea of a command representing future intent, and an event representing a past tense fact about something that has happened, what could a command event possibly mean? And I just have to come back to Alanis and think, Isn't it ironic? You send a command event and honestly, I can't even finish the rhyme. It just makes no sense. There's no world in which a command event is a thing that makes sense to me. I just cannot parse it. These things are just completely opposite things.

And I can only imagine a world in which you would think about a command event as being one in which you've internalized the technical mechanism underlying these sort of topics in the streaming systems. And so everything becomes an event, and then you're starting to think about events as messages, and you've lost the original semantic and logical meaning of what it means for a message to be an event. So, command events don't exist. Don't believe if you read anything that refers to a command event, it's nonsense. Commands and events are distinct logical intentions of different messages in the system.

So, let's wrap up our differences between these different things. We've got message read status tracking. In the queuing system, it's handled for you by the broker. In the streaming system, it's handled by the consumer, by the client library, but in that process which affects scale out. If we have poison message handling, the broker will take care of that for you. But in a streaming system, it's up to the consumer. Can you auto-scale consumer clusters? In a queuing system, yes. And if you're using the right type of queuing system, pretty much you can scale way beyond anything you would need.

In a streaming system, yes, you can, but not beyond the partition limit. You have to think about that partition allocation and reallocation depending on as you hit that scaling limit. Can you order recover from consumer process failure? In queuing systems, yes, automatically. In streaming systems, well, yes-ish. Client libraries do the heavy lifting for you. And it does work, but there's sometimes some delays, depending on detection of node failure and so on. Is there support for this concept of semantic commands versus events in queuing systems? Yes. Bypass the topic, go straight to the queue. In streaming systems, no, you have to apply your own convention and semantics onto the system.

Storage of historical messages. This is an interesting one. Consumers in a queuing world, it's their responsibility. If they want to store a copy of a message, they have to do something, like forwarding into an audit queue or keeping their own database of the history of messages. In the streaming world, the broker will do that for you. And honestly, this is one of the big benefits that a lot of people see when they're just making a decision between these two systems. They like the idea of keeping the events around. They like the idea that they can bring online a new service and re-initialize it from the entire history of events.

This is something that's often more difficult to do in practice. Schemas change. Historical events from years in the past maybe are not going to get processed in the same way. This is not going to be as simple as it seems. And I do encourage you to think about not relying on that, especially if you plan for this system to run for a decade or more. Reprocessing the entire history of events is not going to be something that you ever actually really want to do. And it relates to this, can new consumers access the historical messages in queuing worlds? Not typically, unless you do special work around it, but in streaming worlds, by default, that is the case. You can, of course, configure the streams to expire messages, but this is the thing that happens.

You can start to see here that I'm not saying you can't do any of these things. There are ways to do it. You can do anything that you might want to do with either of these two systems. The question is not can you do it? The question is whether or not the system is finally tuned for it, whether it was designed with that intention in mind, whether it was optimized for that scenario. And hopefully now, you're starting to see the different scenarios that these are playing out for.

I'm going to take a brief tutor now into event sourcing and event stores before we come back to, I guess, the final summary of the recommendations around how to think about these two different systems. So, if queues and streams are about inter-process communication, event stores are not at all about inter-process communication. They're actually about storing the internal state of your system as events, a record of the history of it.

Because in event sourcing, what we're interested in is actually the thing that happens between submitting the order and the order becoming submitted. In demand-driven design, we often refer to this as an aggregate type of entity. There's some nuance there. I'm not going to go into it today. Now, when we think about storing state, traditionally, we often just stored this current state of the system. We would store a row in a database table, and each row represented an individual instance of the entity, order one, order two.

Or maybe over the years you migrate it to NoSQL and you store a blob for order one and a separate blob for order two, keyed by the order ID, and the system allows you to retrieve the current state of the system. In that kind of world, event-driven architecture becomes a side effect of persisting state. You persist the state. This now becomes a fact about the system. You want to advertise that fact, so you publish an event.

In event sourcing, again, you invert that relationship. You actually don't store the state. You actually just store the history of events and you split your aggregates into two halves. One half is responsible for receiving a command and making a decision about what event is the thing that occurred if a command came in, given the current state of the system. And then you just capture that event. But I just said you have to know the state of the system in order to process a command.

And in event sourcing systems, you typically compute that state on the fly as a side effect of persisting the event. So we can think about the difference between storing the state and publishing events as a side effect, or storing the event and computing the state as a side effect. And this is at the heart of the difference between state persistence and event sourcing.

Now, I'm not going to go into much more detail on this. There's some great talks and literature out there online to help you make the decision between these two. But what I really want to make sure is very clear here, is that even though this uses the word event in the same way that other streaming and queuing system uses the word event, it's a very different proposition.

And we're not talking about communicating between processes, we're talking about managing the state of a private system. And we can see this really clearly when we think about streaming systems and event stores, is that Distributed Logs, so streaming systems like Kafka, they're kind of like databases that are optimized for this query. Select star from events where topic equals topic and offset is greater than last offset. Remember, I said we're always requesting with an offset.

Whereas event stores are optimized for select star from events where stream ID equals entity ID. Totally different things. In streaming systems like Kafka, you can't get the history of events that relate to a given entity ID. You can only get the history of events that are a particular type or topic. Whereas with event stores, that's the only thing we want. We want to know the history of all different types of events that occurred to a given entity.

So this different access pattern is fundamental. And I think you can very clearly see now why Kafka as a streaming service is not an appropriate solution for when you're trying to do event sourcing with an event store. So you've got this difference between stream per topic versus a stream per entity instance. So why on earth would anyone think that they should use Kafka as the store in an event sourcing system? Well, maybe it's because in 2017 the Kafka business published this blog post.

The post discusses event sourcing in the context of Apache Kafka, examining the need for a single source of truth. And it talks about using Kafka as the source of your events in an event sourcing system. I think this led to a lot of complexity in systems that had to deal with trying to use this service in a way that it wasn't intended for, and resulted in having to pre-compute and store the state for every single message, which obviates some of the benefits of event sourcing in the first place.

So, isn't it ironic it's an event streaming log when your state's out of date? Just don't do it. It's not what it's for. Thankfully later, they did go back a little bit and wrote another post about event sourcing, and actually ended up describing this scenario where we say, "It's a mistake to assume that every event stored in an event source system is also emitted to Kafka." They started to acknowledge that your log of events as part of your event sourcing is separate, and some of those events will get published into Kafka and made visible to other processes. I do think it's a bit of a shame here. They talk about domain events and they separate those from integration events. And I'll talk about it a little bit in a moment why I think that's a bit of a mistake.

And it's because if we go back to this Logical View, our event storm outcome where we have the flow of commands and events through the system, and then we added these things into bounded context. We make a decision about what parts of the business process are going to be the responsibility of different processes and different bounded contexts in your system. We need to be actually more clear than this. And what I want to do here is show that some of these events exist on the boundary of the bounded context. They flow from one bounded context into the other.

So we can distinguish between domain events that are private domain events and domain events that are public domain events. They're all domain events. They all represent important moments in the domain, but some of them are private to the bounded context and some of them are public to the bounded context. They're part of the public API of the context, and other contexts can take a dependency on them and become coupled to them. And I think this is a far more important useful framing than a domain event versus integration event. Public and private is a useful way of thinking about it.

So finally, we can get to the end of this, some recommendations around when to use which, which kind of events relate to Kafka topics or queuing systems? Let's think about each of these types of events we've now seen here. We'll start with private domain events. So we're talking about events that exist within a bounded context and are not intended to flow out to other parts of the system.

And they probably capture lots of data because we're doing the actual storage of the state of the system on the history of these events. We want to use the events as the source of truth and the historical record, use an event store. If you're in the .NET world, something like Marten is a great example of a library that uses Postgres to give you an event store as a private storage of the state of your bounded context. And truth be told, none of the other services in your system, none of the other bounded contexts should even be aware of whether you're doing event sourcing or state resistance. It's a private decision of that bounded context.

If the purpose of your domain event is actually to trigger the next step in an internal workflow, some sort of business process choreography, then we need to think about this a little bit differently. Each event is critical. It's important that the event is picked up by the next step in the process and flows through. In this case, you might want to reflect this event out onto a queue-based pub/sub system.

It will be flowing back into the same bounded context. We will typically store the event in the event sourcing system. There'll be some sort of a subscription mechanism in the event sourcing system to allow you to hook into that and reflect the event out into the interprocess communication. And then you will be able to receive that event back into the same logical context and execute the next step in the business process.

The reason I recommend going out of process in this moment is because you don't want to couple any sort of complex processing to the persistence of that event sourcing store. It's important then to, if you want to get the benefits of retries and dead letter queue handling, and scaling out the processing, to emit that event out into process communication and allow it to be handled again.

If we're talking about private events that we actually want to get some sort of analysis on, internet of things, click streams, anomaly detection, moving average threshold crossing, statistical aggregation over time, we typically find that the individual events are less important than the overall stream, the statistical averages of the stream, in which case this is what streaming with a Distributed Log is for.

You can use a similar message to omit those events onto the streaming system and hook it into the appropriate analysis tools to give you that analysis of the stream over time, or streaming analytics as they call it. Again, the producer and the consumer belong to the same bounded context, and you might use something like ksqlDB or Flink to give you that processing.

Let's now talk about public domain events. I do apologize, I realize we've just gone over time. I'm not sure if it's 'cause it's 2:00 in the morning with the technical issues we had at the start, but I'm running a little bit late. So, thank you for your patience if you're bearing with us just towards the end. We're nearly there.

Let's say we have a public domain event where we actually do want it to go out to other services. Typically, I recommend here that we don't put a lot of data on this. The ideal for me is just to have the ID of the order. We want it to be lightweight, 'cause the smaller the contract... Remember, these are public now, so they've become a contract that other services are dependent on.

The smaller the contract, the less risk of change propagation, the less risk that something you modify in your shopping cart is going to bleed into having to update your subscribers so they can handle the schema changes or handle the differences in that contract. So, having a lightweight message is a really good heuristic for reducing coupling between these services. Sometimes I get the pushback here, "Isn't this a clickbait event?" Oscar Dudycz is a real expert in EDA and he coined this term, clickbait event.

I think that it depends on how you use this type of event as to Whether it becomes a clickbait event. When Oscar was talking about clickbait events, what he meant was an event that's being published from one service to another with just an ID, in which the downstream service pulls back into the upstream service to retrieve the data that it needs. So, you might look at that small event and say, "Oh, this is nice and loosely coupled.

It's a very lightweight event," but actually the downstream service still needs the data. So, it's going to be making an API call back in and retrieving that data. So you think you've reduced the coupling, but you've just moved it into that pullback API call. This is a terrible pattern, and this is actually what Oscar is referring to with the clickbait event when he's advising against it.

But there's a way to avoid this situation. And it comes back to looking at the event storm and really thinking deeply about the situation where we're trying to think about, "Where are we capturing the data versus where are we using the data?" So again, in this simplified example, I actually glossed over an important step. And so we have our shopping cart, we've added the items to the cart. We never captured the process of capturing the shipping details.

So let's say we do enter the shipping address and we have the shipping address set before we submit our order. And then we can start to see, "Well, why would we capture that data in the shopping cart just so that we have to propagate it downstream to the shipping service, which is the service that actually needs that data?" An alternative design here is to split the shopping cart's responsibilities and to inject the shipping service or the shipping bounded context between these two steps.

And we can see that it's totally fine for our bounded context to be discontiguous in the work stream of the business. We can have the shopping cart capturing the details of the items. We can have the shipping service capturing the shipping address, the shopping cart handling the actual submission of the order and the workflow that proceeds from them. And because the shipping service captured the shipping address, it now has the data already that it needs in order to process the shipping of the order.

And this can be a very effective mechanism to reduce the coupling between your services, by thinking about the distinction between the capturing of the data and the usage of the data. And so this can become definitely not a clickbait event if you have a shopping cart database that stores the shopping cart details and a shipping database that stores the shipping details. And all you're doing when you publish that event is telling the shipping service to use the data it already has to process the next step in this order. So, this is recommended, and it helps you use these lightweight events without them becoming that kind of clickbait event.

So if we're talking about a public event that's there to allow the subscribing bounded context to react and execute the next step in a workflow, each event is critical, it's important that they happen and they're processed eventually, then I recommend messaging with a queue-based pub/sub system. And the producer and consumer in this case are different bounded contexts.

Now we have another interesting scenario, which I call a domain event with a question mark, because sometimes you do end up needing to publish an event with a lot of data on it. Maybe you have a customer data record, and the customer has updated their address, and you need that address in many different downstream systems. It's a large complex data structure. There is risk of change affecting the downstream subscribers if it's not stable in your domain.

And this is an example of what's commonly called event-carried state transfer data sharing. We're actually carrying the state of the system on an event so that we can transfer the state of this entity into other systems. I think that it's really important to recognize that this is actually not a domain event. This is not probably an event that would've appeared on your event storm.

This is not something that your business stakeholders actually care about. What it is is a mechanism that you are using as a technical design decision to get the data from a place that you've decided it should live to the place that you have to use it, which is a technical reason for moving this data from one place to another. It's not a domain-driven reason for moving the data.

So I think this is actually a really good example of where we should use the term integration event, because it's about a technical integration between different components of your system as a consequence of the design decisions that you've made. It does not reflect an intrinsic part of the domain. So, streaming with a Distributed Log can actually be really effective here, because often in these cases, individual events are not that important. Later events are actually more up-to-date. So you actually prefer to skip over events and just get the most recent one if that gives you the most up-to-date information.

Typically, the producer, the consumer are belonging to different bounded contexts. And there are services like Kafka Connect that actually make it very easy to receive these events and project them into the current state of the system, in what Ian Cooper, is a guy I admire deeply and is very experienced with event-driven systems, calls a forward cache. It's a way of giving the downstream systems access to the data that's owned by the upstream system as a way of forward populating this cache so it's ready to go in a local storage that's always online and available to the downstream system, without it having to call back and add API load onto the upstream system. And although I personally would prefer avoiding this in a system design, if you end up in a system where you need to do something like this, this is the way that I would recommend doing it and the way that I would recommend thinking about it.

Because if you think about what we're talking about with events here in systems, we can't forget the principles that we learned with databases of encapsulation. If I was to show you a system diagram that looked like this, where you have a shopping cart service with its own shopping cart database, and you have a downstream payment service that is reading directly from that shopping cart database, you would probably appropriately recognize that this breaks encapsulation. This is a bad idea.

The whole microservices movement was about trying to stop doing this kind of thing and mediate access to data through well-defined APIs to encapsulate change and allow services to own the business rules and logic associated with their data. So if you have a private domain event topic in your system, is it not just the same thing if you allow downstream services to subscribe to that event?

Just because it's an event doesn't mean you should definitely let all downstream services subscribe to it. If you've modeled it as private and you're storing it in your event store, as soon as you let other services subscribe to that event, you've created coupling that may be invisible to you and may cause you problems down the line. If you think that you can evolve this, you're going to end up breaking these downstream services.

So, I think it's really important to be very clear about the distinction between private and public and to treat those public events as contracts. But this does become harder with event-driven systems, because with a database, you put the data in the database and you just block access to it. You say, "Only this service can access this database. If you want the data, come in via our API."

But once you start pushing events out onto public brokers, some of those events you might want to keep private. Some of them you might want to be public, but they're all sitting in the same broker, this mediated technology layer that sits between all these services. How do you make sure that the downstream services don't accidentally subscribe to or start reading from these private events?

Now, there's a few different mechanisms that I recommend. If you're using the same programming language across your whole system, well, use the coding language features. Put your private events into your application assembly, mark it as an internal record, and just don't let the downstream services even be aware that it exists. Put the public events into a dedicated contracts assembly, which you can publish via NuGet or other mechanisms so that your downstream services can be aware that this event exists and start to subscribe to it. And so the language features can give you support for this kind of thing.

If you are not using the same languages across your system, then you might be using something like AsyncAPI to document your event APIs. Just make sure that your private events don't end up in your AsyncAPI documentation, again, so that the downstream services just don't even become aware that it exists.

Maybe you're seeing some documentation of your event-driven system with a tool called EventCatalog, which I do highly recommend. Dave Boyne from the UK has been working on this. A lot of businesses are getting a lot of value out of it. He doesn't currently support this, but he told me the other day that he's adding currently support for explicitly marketing events as private versus public. So hopefully keep an eye on that and be coming there soon.

So let's just wrap up with a brief summary of the heuristics here. So, we're talking about private, sorry, storing events as state. Private events go into the event store, and you shouldn't really be doing public events as the mechanism of state storage because state should be private to the bounded context. We're talking about business process orchestration, whether it's a private event that flows back into the same bounded context or a public event that goes to other bounded contexts.

Queue-based pub/sub, I think, is the way to go for any kind of business process workflow management because of this built-in support for retries and dead letter queues and auto-scaling. It's just well-suited to this scenario. We're talking about streaming analytics and processing many, many messages and aggregating across them, and being interested in the statistics or the aggregation of the overall stream, rather than doing something specific in response to an individual message, then that's what our Distributed Logs are for. That's what streaming services are for.

If we're talking about forward caching, then again, Distributed Logs can work quite well. There's mechanisms around them to help you. And you have to be aware that even though this might be a lot of data that previously was private, as soon as you're putting it into a forward cache that's available to another bounded context, it's become public. It's now a contract. You need to manage it as such.

So, there's quite a lot in this talk and we have gone a little bit over time. I do apologize. But please, hopefully this resonated with you, and please don't let this be the good advice that you just didn't take. If you're interested, there's a GitHub with some code examples in C# that illustrates the differences between these different situations. There's some unit tests that will provoke the same scenario, whether that's a failed message or a failed processing node.

And we'll send these messages through three different types of brokers, a queuing broker using RabbitMQ and two different versions of Kafka, the low-level consumer Kafka and the high-level consumer Kafka, so you can actually see in the test assertions what the expected outcomes are. So check this out on GitHub. It's called Queues versus Streams, I think.

A lot of this material is built from a training course that I offer. It's called Imagine the Future and Build for the Now. It's got a lot more than just queuing and streaming services in it. There's a live session coming up online in October. The whole agenda is related to Christian's 4+1 views. We talk about the Logical View, we do event storming.

We go into the Process View. We talk about inter-process communication, which is where a lot of the content from this talk comes from. We talk about the Developer View, and how you organize your source code. And the Scenario View, which we didn't touch on today, but we can start talking about user story mapping and prioritizing your implementation planning and how that relates to your architectural thinking.

So I have a lot of fun presenting this talk. I've done it at conferences around the world. This is going to be the first time it's offered publicly in an online forum, so please do check that out if you're interested. And if you'd like to follow up with questions to me privately after this, please do reach out. My website's there, LinkedIn, Bluesky, Mastodon. Very happy to chat. Love talking about these topics. Welcome you to reach out. And thank you very much. I'll leave it there with some details about Particular that maybe Hazel would like to share with you.

Okay, awesome, Chris. We have a couple of questions from the chat that I was hoping we could address real quick. So the first one is from Ashif. "How do you avoid bottlenecks caused by the database when competing consumers are required to access the database when processing messages?

Specifically when there's a burst of messages arrives in the queue and then scaling out the consumers would start opening many concurrent connections to the database? One approach that they've seen is processing these messages in batch and limiting the consumer to one connection to the database." What's your thoughts on that?

Yeah. So it's an interesting scenario. I mean, it'd be great to dig into how that was causing a problem, because typically with a queuing system, one of the advantages of it is that if there's processing constraints in the downstream system in the consumer, then the messages just sit in the queue. You're not forced to process these messages on any particular rate. It's up to the consumer to start processing the messages as it sees fit.

So if it receives a message and opens a connection to the database, it doesn't have to process the next message concurrently. So, what this suggests to me is that you've scaled-out your web tier to a far greater capacity than you've scaled-out your database tier. And so there's actually, within the consumer, a mismatch in the processing capacity of the, not the web tier, I shouldn't say, but the compute layer from the data layer.

So, one of the ways that you can do this is actually use the queue to your advantage and limit the processing capacity of your compute tier so that it matches the processing capacity of the database tier and just reduce the concurrency. Just let those messages sit in the queue until the consumer has completed processing them and then picks the next message up off the queue.

That would be the way that I would deal with it because that gives you the opportunity to handle things at the maximum capacity, but no more. If you're ending up in a situation where you open up way too many concurrent connections to the database and the database is your bottleneck, well, you've just basically pushed the bottlenecks for there. Ah, you're talking about Azure Function app message handling instances spinning up dynamically.

Trying to recall with Azure Function App instances where you can put a concurrency limit on it. I'm pretty sure with AWS Lambda, the equivalent, you can do that. I can't recall off the top of my head with Azure Function app message handlers whether you can. So, other possibilities are looking at scaling up your database to handle it. But I think that moving to batch processing to... So you're effectively building your own private queue with memory when you've got a queuing system right there.

I think that's probably about the extent to which we can give an answer in this setting. Yeah. So you've ended up with 10 concurrent instances. I'd be looking for ways to limit that concurrency. One thing that I've done where we had a downstream system that was very heavily intensive, it was calling a downstream API and overloading it, is that we put rate limiting in place. So we put just a very simple lock mechanism that would that, for each instance of the node, limit how many messages it would process at a given point in time.

If it receives a message and it had not exceeded the timeout between messages, it would just push it back onto the queue. But I'm not sure how you'd think about that as your function app instance, if it's spinning them up dynamically. I think my first instinct would be looking to ways to limit the concurrency so that the compute layer capacity matches your database layer capacity, and then let the queue do the work for you of keeping track of the messages that are not yet processed.

Because I'd be really worried about receiving a batch and then a failure in the middle of that batch. Do you reprocess the whole batch? Where do you keep track of the status of the individual messages? You're kind of recreating your own queuing infrastructure. I'd be nervous about some of the consequences of that. Sorry, Ashif, I'm not sure we can get into more detail, but if you want to reach out on LinkedIn, I'd be happy to have a more in-depth discussion with you at another time.

Fantastic. We also have a question from William. "In a stream-based system, when a failure occurs and you skip that offset, can you ever eventually go back to reprocess that failed message at the offset where it failed? Do any of the high-level APIs allow you to track which offsets were failures or not?"

Not the default Kafka ones, no. The interaction is just giving the message from this offset, giving the messages from this offset. So, you can go back and retrieve a message at a particular offset and you can request just one, but then you have to manually make sure that you reset the offset that's being tracked back to the current head of the stream, if that makes sense.

Plucking messages out of the past and just getting that one message is not really what it's optimized for. And certainly, none of them allow to natively track which offsets were failures or not. There are some other libraries out there, mostly in the Java world. I do apologize, I don't know them on the top of my head, but if you're searching in this space for extra libraries to help you with this kind of situation, that would be the way that you would have to handle it.

And so what those libraries do is that they manage the state of these messages independently. And a common pattern is to have dedicated topics that are retry topics or dead letter the queue topics related to the initial topic. And for when the failure occurs, to put the message into the retry topic and then have a separate process that's monitoring that, to then reprocess it or push it back into the original topic.

But again, because you're storing the whole history of the messages, you can end up then with multiple copies of the same message in the history. And those retry messages are going to sit in that retry topic forever as well, unless you specifically configure it to expire those messages. You can see why I say it's not really what it was designed to do. Almost everything that you have to do to deal with failed messages, and make sure that they get processed in a streaming system under all sorts of failure modes, ends up being a lot of workarounds and complexity on the client side.

And I have spoken to some people who said, "Look, Chris, you're full of shit. You don't know what you're talking about. I've used Kafka for a decade in an environment where it was really important that we processed all these messages." I'm like, "Okay, sure. Tell me more about how you did that." And he goes on to explain this very elaborate contraption that helped them deal with all these things. And I'm like, "If you had a queuing system, you would've got all of that for free." So, I think it's just what they're optimized for. It's what they're oriented around, is what I've seen.

Awesome. Thank you so much, Chris. So, we have some upcoming events for Particular that I'd like to let everyone know about. We're here at VS Live in Redmond right now, but if you're on this webinar, you're probably not here at the conference with me as well. No worries, though. Particular Software will be pulling double duty in September. From September 14th to 18th, we'll be at NDC Oslo and at VS Live in San Diego.

At NDC Oslo, Laila Bougria will be giving her talk Changes is Inevitable: Versioning Event-Driven Systems. And William Brander will be presenting Indiana Jones and the Temple of Legacy Code at VS Live in San Diego. Hazel Bohon, that's me, will be presenting Microservices for Pragmatists if you're able to. Both conferences are really wonderful and I would highly recommend you come out. The agenda for both conferences is packed with some really great talks that you won't want to miss.

Here's for a full list of upcoming events where we'll be speaking. Go to our events page at particular.net/events, or you can scan the QR code on this slide right now. Thank you all for coming to today's webinar. On behalf of Chris Simon, this is Hazel Bohon saying goodbye for now, and see you at the next particular webinar.

It's like 10,000 streams when what you need is a queue

🔗Why attend?

🔗In this webinar you’ll learn to:

🔗Transcription

About Chris Simon

Additional resources