How do event sourced systems make sure the read model is only updated once?

Question

Currently looking into Event Sourcing in a Microservices architecture and I've found this article. It describes a possible implementation of CQRS and event sourcing.
If the logic dealing with updating the read model and creating the events are both implemented in the same (scalable) service, how does this architecture make sure that an event changing the read model is only handled once?

To clarify with the example used in the article: If the User Management Service is scaling, leading to multiple instances running at the same time, how can it be ensured that the logic doesn't add the same user twice?

One solution I can think of is completely splitting every Microservice into one read- and one write-service and only scaling the read service, but that doesn't sound optimal.

Answer 1

If the User Management Service is scaling, leading to multiple instances running at the same time, how can it be ensured that the logic doesn't add the same user twice?

There are a couple of different answers here.

A good starting point would be Marc de Graauw's Nobody Needs Reliable Messaging. The basic idea is that the multiple messages have the same business semantics, and therefore consumers can reject duplicates themselves.

That in turn means pushing backwards through your protocol; if we have two copies of an HTTP Request, possibly separated in time, that attempt to create the "same" user, then two instances of the User Management Service handling those requests should end up trying to add semantically equivalent events to the store.

With that property in place, any given consumer can use the semantics of the message to eliminate duplicates.

Event store implementations can help with Conditional Publish. For example, Event Store achieves this by supporting an expected version parameter in the write command.

In a race between competing producers, the two writers will be competing to write to the same expected version of the stream; one producer will succeed, and the second will fail -- the second producer then knows that its locally cached representation of the stream is out of date. It can then refresh its cache, and try processing the message again.

In other words, writes to the event collections are achieved by compare and swap of the tail reference, rather than by "append".

As far as I can tell, in 2017 Kafka doesn't support conditional publish. The exactly once delivery feature in 0.11 doesn't appear to handle that case.

Multiple processes writing to the same event partition may not be what you want. Reasoning about the behavior of a single authority is so much easier. Instead of having multiple instances of the User Management Service sharing the authority to write to a single stream, you might be better served by creating multiple streams, each with a single authority (essentially, each distinct stream has its own leader election).

I don't think I understood the last part 100%, though; would that look somewhat like this: i.imgur.com/b0C2xNV.png?

Yes, in the sense that each user service has its own topic (book of record) for output events. But also, you want to be sure that all events related to a specific entity in your model get written to the same topic. So there would be a piece of logic responsible for ensuring that each command gets handled by the authoritative instance of the user service for that entity.

Answer 2

Apache Kafka includes a feature for automatically managing consumer partition subscriptions, so that only a single consumer in a particular consumer group (such as the user read model updater consumer group) is subscribed to a particular partition - it balances the partitions across consumers automatically (it uses ZooKeeper behind the scenes to manage this). So if you scale the user management service, the new instance will add itself to the read model updater consumer group, causing the partitions to be re-balanced across the different consumers. If you remove an instance, its partitions will be added to remaining consumers in the group. This should ensure that only a single process and thread will be writing to a particular read record at a time, preventing any concurrency problem.

The write side is trickier in Zookeeper. There are two problems to overcome:

1. Zookeeper can't handle having a separate stream per aggregate (mostly due to the partition management, which is handled per-stream, so doesn't scale past 1000s of streams). This means that all aggregates of a particular type (User, say) need to share a single stream.
2. Zookeeper has no built-in means of locking a stream for writes (either using optimistic or pessimistic locking), so you need to ensure that events for a particular aggregate are committed serially yourself.

The first issue means you can't really load a single aggregate by loading its events (since you can only load events e.g. for all Users, which could be millions of events), so instead you need to have consistent snapshots of all aggregates (either in memory, or in a store, depending on scale and allowable startup latency). Every time you publish an event, you immediately save the snapshot (and retry until it works, or your process dies). At startup, in case of a previous failure, you need to work out the event number of the last written snapshot, and reprocess events from there to fill in any missing snapshots. You can cache snapshots in memory of course.

To handle the second issue, one option is to ensure only one thread+process handles a particular aggregate. The simplest way is just to have a single writer, as you mention. Alternatively, you could post commands (as opposed to events) to another Kafka topic, and use partition management again to allocate partitions to command handlers, safe in the knowledge that all commands for a particular aggregate will hit the same consumer, and hence be processed serially. The command handler can potentially post a reply back to a topic specified in the command metadata so that the API gateway can return the command result when it's available.