Event streams
In short: An event stream such as Kafka is a log where each event is written once and every system reads it at its own pace.
Problem
One order, four systems
Customer 7 places an order for 12.40 €. The warehouse has to pack it, the invoices team has to issue an invoice, the e-mail service has to send a message, and the analytics have to count the sale.
If the order code calls the four systems one after the other, a mail server that does not answer makes the whole order fail. Or the order goes through and the e-mail is lost. Every new system (say, loyalty points) means a change in the order code.
The orders table keeps what the order is now. It does not keep who already knows about it or what happened before. You need a place where the shop says “the order was placed” once, and everyone who cares reads it when they can.
The idea
A log of events
An event is something that has already happened: “order 61 was placed”, “order 61 was paid”. The shop writes it at the end of a log and forgets about it. The warehouse, the invoices and the e-mail each read the log at their own pace. If the e-mail service is down for an hour, it carries on from where it stopped, and nothing is lost.
A log like this is an event stream. Kafka keeps it in topics, for example orders. A topic is split into partitions, so that many machines can write and read at once. Each record gets the next number in its partition: the offset. The numbers only go up.
produce writes an event and tells you which partition it went to and under which number. read shows everything: p1:0 is partition 1, offset 0. In the picture each row is a partition, the cells are the records with their numbers (under the number is the end of the key), and the dashed cell is where the next record goes.
Records are not changed or deleted: commands like update and delete give an error. If something is put right, that is a new event.
More precisely, if you are curious
The simulator commands are a more readable version of the Kafka tools (kafka-topics, kafka-console-producer, kafka-console-consumer, kafka-consumer-groups):
create topic orders partitions 3: a new topic with 3 partitions;topicslists the topics;produce orders key=customer-7 {"order":61,"total":12.40}: writes an event; the key is optional;read orders [partition 1] [from 0]: shows the records and remembers nothing;consume orders group billing [as billing-2] [max 5] [from-beginning] [no-commit]: reads as a group and saves how far it got;offsets orders: the end of each partition and the position of each group;describe group billing: the position, the end and the lag of the group in each partition;groupslists the groups;reset group billing to 0(orearliest,latest): moves the position of the group.
no-commit exists only in the simulator: it shows a consumer that has read but crashed before saving its position. The simulator reads partition by partition (first 0, then 1, then 2). A real consumer fetches batches from several partitions at once, and the order between them is not defined.
See it
Order is lost between partitions
Without a key the records are spread over the partitions (the simulator takes them in turn: 0, 1, 2, 0). Order 61 is placed, paid and shipped. The e-mail service reads:
The e-mail service sees “shipped” for order 61 before “placed” and sends “your parcel is on its way” for an order it does not know about yet. Order is guaranteed only inside one partition.
That is what the key is for. The program that writes computes a number from the key (a hash) and takes the remainder after dividing by the number of partitions. The same key always gives the same partition. The simulator’s hash is the sum of the character codes: for customer-7 it is 982, and 982 divided by 3 leaves 1, so partition 1. Kafka uses a different hash (murmur2), but the idea is the same.
The key is whatever needs its order kept: the order number for the events of an order, or the customer if the order of all their orders matters.
Try it
The events of one order
The topic orders has 3 partitions. Write the three events of order 61 in order (placed, paid, shipped) with the key order-61, so that every consumer reads them in that order.
Solution
produce orders key=order-61 {"order":61,"event":"placed"}
produce orders key=order-61 {"order":61,"event":"paid"}
produce orders key=order-61 {"order":61,"event":"shipped"}
Three times produce orders key=order-61 …. All three go to one partition (1) under the numbers 0, 1 and 2, and every consumer reads them in order. Without a key they go to three different partitions and the order is lost. The key keeps the order you write them in: if you write “shipped” before “paid”, that is how they will be read.
The idea
Consumer groups
Each system reads as a consumer group: billing, email, warehouse. The group remembers how far it has got in each partition. After reading it saves that position (a commit), so it carries on from there even after a restart. Two groups do not get in each other’s way: each one gets every event.
billing has read everything, and email only two records. The difference between the end of the partition and the position of the group is the lag. In the picture each group has its own row under each partition: the coloured part is read, and ▲ is where it will read next. A growing lag means the group cannot keep up.
Inside one group the partitions are shared. If invoicing is slow, you start a second copy with the same group, and each copy reads its own partitions:
That is why the number of partitions is the upper limit: with 3 partitions a fourth copy sits idle.
A new group without from-beginning starts at the end: it sees only the events from now on. from-beginning applies only to a group that has no saved position yet.
The analytics team starts today and wants all the orders so far. Read them with a new group analytics.
Solution
consume orders group analytics from-beginning
consume orders group analytics from-beginning. The new group has no position and starts from the first record. Without from-beginning it starts at the end and reads nothing. And with the group billing the analytics would take the invoices’ position: that group has already read everything.
See it
Two invoices for one order
The group saves its position after it has read. If the consumer crashes between the processing and the save, after the restart it gets the same events again:
Orders 64 and 61 come twice. If billing sends an invoice every time it reads, the customer gets two. That is why the consumer has to be idempotent: applying the same event a second time changes nothing. The most common way is to let the database refuse the second one:
UNIQUE on the order number and ON CONFLICT DO NOTHING make the second event harmless.
More precisely, if you are curious
“Exactly once” is hard. The other order (save the position first, then process) gives “at most once”: on a crash the event is lost. So the usual choice is “at least once” plus an idempotent consumer. Kafka has an idempotent producer and transactions, which keep “exactly once” for reading and writing inside Kafka itself. An e-mail that has gone out cannot be taken back by any transaction.
The idea
From the start again
The analytics report was wrong: it forgot the delivery fee. The fix is ready, but the old orders have already been read. The log keeps the events, so the group can go back and read them again (replay):
reset group analytics to 0 moves the group’s position to the start of each partition, and the next read goes through everything again. The other groups notice nothing. from-beginning does not help here: the group already has a position, and the position wins.
Replay is safe only for an idempotent consumer. A report that is computed again is fine. Invoices that are sent again are not.
Records still do not stay forever. A topic has a retention period: events are deleted when they get too old (in Kafka, after 7 days by default) or when they go over a set size. Age deletes them, not the consumers: what everyone has read stays until then, and what nobody has read is gone after that. The simulator does not model retention.
More precisely, if you are curious
In real Kafka kafka-consumer-groups --reset-offsets needs the group to be stopped and --execute, otherwise it only shows what it would do. There are also compacted topics (log compaction): for each key at least the latest record is kept, and older records with the same key are deleted.
The idea
The state is the sum of the events
In the products table the stock of white cheese is one number:
The number is 18, but it does not say why. How many came in, how many were sold, was anything returned? UPDATE writes over the old value and the history is gone.
Event sourcing keeps the events as the truth, and the state is computed from them: the stock is the sum of all the changes. The same in SQL, with the stock after each event:
And this is how it looks in the stream. The key is the product, so all its events are in one partition and in order:
The events give you the history for free: who did what and when, the stock on any past day, and a new calculation over the old events. The cost is that the current state means going through all the events. That is why programs also keep a ready state (a snapshot) that they update from the stream. The events stay forever, so their format has to change carefully too.
Exercise
Rewind, a second copy and a correction
1. The groups billing and analytics have read everything in orders.
The analytics calculation is fixed. Move the group analytics back to the start of orders so that it reads everything again. Do not touch billing and do not read yet.
Solution
reset group analytics to 0
reset group analytics to 0 (or to earliest). offsets orders shows analytics at 0 in each partition and billing at the end. consume … from-beginning does not help: the group already has a position. reset group billing to 0 would send the invoices again.
2. billing has read everything, and then order 65 arrives (in partition 2).
Start a second copy of the invoicing called billing-2 in the same group, so that it takes some of the partitions and reads what is new in them.
Solution
consume orders group billing as billing-2
consume orders group billing as billing-2. The two copies share the partitions: billing-1 takes 0 and 1, and billing-2 takes 2 and reads order 65. group billing-2 is a different group: it would get every event separately, and every customer would get two invoices.
3. The topic stock has the events of white cheese (product 7): a delivery of 24, sales of 4 and 3.
The customer from order 61 returned one cheese. Add the event {"product":7,"change":1,"why":"returned"} so that it is in the history of the cheese, after the sales.
Solution
produce stock key=product-7 {"product":7,"change":1,"why":"returned"}
produce stock key=product-7 {"product":7,"change":1,"why":"returned"}. The event goes to partition 2 with the others for product 7, under number 3. Without a key it goes to another partition and its order with the sales is lost. Changing the old sale is not possible: records are not touched, and a correction is a new event.
The idea
When a stream, when not
An event stream is good when many systems have to learn about one thing, each at its own pace, and when the history is worth something: orders and payments, stock movements, clicks and user tracking, server logs. It is not good when you need an answer right away (“is there cheese in stock?”, “what is the total?”), searching, or changing a record. That is what the SQL database is for, and it usually sits next to the stream and is filled from it.
More precisely, if you are curious
Kafka is the most widely used: topics, partitions, groups, and the positions of the groups are kept in Kafka itself (in the topic __consumer_offsets). Redpanda speaks the same protocol as Kafka, so Kafka programs and tools work with it. RabbitMQ is first of all a queue, where a message disappears once it is acknowledged. Its streams are a separate kind of queue: a log that can be read from any position and read again. Amazon Kinesis Data Streams calls partitions shards and by default keeps records for 24 hours.
Inside one browser page the same idea is an event bus (publish/subscribe): a button announces “added to the cart”, and the counter and the total subscribe to it. The difference is that a bus keeps nothing: whoever is not listening at that moment misses the event. A log keeps the events and lets you read them again.
Test