CQRS and Event Sourcing with Lagom

Lagom is the new microservices framework from Lightbend (formerly Typesafe, the company behind Scala and Akka). The framework and the concepts behind it are heavily based on CQRS (Command Query Responsibility Segregation) and ES (Event Sourcing). This dictates how state is handled and persisted internally.

In this article I will describe the basics of Lagom and then look more closely at the concepts of CQRS and ES in combination with the framework.

Lagom, The Framework

The philosophy behind Lagom is that it

• has to be distributed
• has to have asynchronous communication
• has to support high development productivity

These ideas dictate how the framework is built. The goal is to develop services on top of Lagom which are very small (in lines of code) and compact. Certain conventions make it straightforward to let the services communicate asynchronously. To give an example of this:

1ServiceCall<CreateCustomerMessage, Done> createCustomer();
2ServiceCall<NotUsed, Customer> getCustomerByEmail(String email);
3ServiceCall<NotUsed, String> getCustomerAverageAge();
4 
5@Override
6default Descriptor descriptor() {
7   return named("customer-store").withCalls(
8           pathCall("/api/customer/average-age", this::getCustomerAverageAge),
9           restCall(Method.POST, "/api/customer", this::createCustomer),
10           restCall(Method.GET, "/api/customer/:email", this::getCustomerByEmail)
11   ).withAutoAcl(true).withCircuitBreaker(CircuitBreaker.perNode());
12}

Three interfaces are being defined here. Because getCustomerAverageAge is a ServiceCall with NotUsed as first generic parameter, it will be automatically generated as an HTTP GET request. A ServiceCall with an object as first parameter and Done as second type will turn this automatically into a POST (even though the type doesn’t have to be explicit within the restCall method. This shows it’s possible with minimal code to define RESTful interfaces that internally are handled asynchronously.
Besided CQRS and ES some other important concepts are applied, such as immutability of objects, design-driven APIs and polyglot programming. Java as well as Scala are supported by the framework APIs, but by using RESTful APIs with JSON data, communication with other services has been made easy.
As the Lagom framework is developed by Lightbend, the technology it is based on should not come as a surprise. Akka, together with Akka Streams, Akka Persistence and Akka Cluster constitute the fundamentals and take care of communication and storage of data. Play is integrated for creation of the RESTful interfaces and for configuration of the framework. Slick is used as ORM framework, where SQL calls are also handled asynchronously. Lastly, ConductR takes care of deploying and scaling the application in production environments.

Some other noteworthy libraries are Logback (logging), Jackson (JSON serialization), Guice (dependency injection), Dropwizard (metrics) and Immutables (immutable objects).
The focus on immutability, non-blocking APIs and a strong presence of the CQRS and Event Sourcing concepts makes the biggest difference when comparing it to frameworks like Spring Boot. Moreover, Lagom is a much compacter framework and offers less functionality. For example, interfaces for queueing are not there and would need work to add and configure. In general Lagom prevents you from having to touch the underlying layers of the framework, but for any more advanced requirements, it will be essential to know and learn about these layers.

Persistence in Lagom

By default Lagom uses the Cassandra key-value store for persistency. As of version 1.2 it is also possible to use a JDBC store, where the principles and APIs are more or less comparable. Later we will dive into using a JDBC store more specifically.
Storing of data works by implementing the PersistentEntity abstract class (a code example will follow later). The PersistentEntity corresponds with the Aggregate Root from the Domain Driven Design concepts.

Every PersistentEntity has a fixed identifier (primary key) that can be used to fetch the current state and at any time only one instance (as a “singleton”) is kept in memory. This is in constrast to JPA, where multiple instances with the same identifier can exist in memory. To add to that, with JPA only the current state is usually stored in the database, whereas Lagom stores a PersistentEntity with its history and all events leading to the current states.
In alignment with the CQRS ‘flow’ a PersistentEntity needs a Command, Event and State. All interaction proceeds by sending Commands to the entity, followed by either an update being executed, or by a response that contains the requested data. So even the querying of the current state is handled by sending Commands.
In case of a change, the Command will lead to an Event that will be persisted. The Event then again results in the State being modified.

Fig 1: CQRS Command, Event, State flow

The next listing shows an example Command for adding a new customer.

1public interface CustomerCommand extends Jsonable {
2 
3   @Immutable
4   @JsonDeserialize
5   public final class AddCustomer implements CustomerCommand, CompressedJsonable, PersistentEntity.ReplyType<Done> {
6       public final String firstName;
7       public final String lastName;
8       public final Date birthDate;
9       public final Optional<String> comment;
10 
11       @JsonCreator
12       public AddCustomer(String firstName, String lastName, Date birthDate, Optional<String> comment) {
13           this.firstName = Preconditions.checkNotNull(firstName, "firstName");
14           this.lastName = Preconditions.checkNotNull(lastName, "lastName");
15           this.birthDate = Preconditions.checkNotNull(birthDate, "birthDate");
16           this.comment = Preconditions.checkNotNull(comment, "comment");
17       }
18   }
19 
20}

How to implement a service (the interface of which we saw in the first listing) and send a Command to an entity is shown in the next listing.

1@Override
2public ServiceCall<CreateCustomerMessage, Done> createCustomer() {
3   return request -> {
4       log.info("===> Create or update customer {}", request.toString());
5       PersistentEntityRef<CustomerCommand> ref = persistentEntityRegistry.refFor(CustomerEntity.class, request.userEmail);
6       return ref.ask(new CustomerCommand.AddCustomer(request.firstName, request.lastName, request.birthDate, request.comment));
7   };
8}

As you can see, the PersistentEntityRef is fetched by using a combination of the type and the identity / primary key. The reference is an instance that you can interact with by sending Commands.

The CreateCustomerMessage implementation (not shown in any listing) is comparable to the AddCustomer implementation from the second source code listing, but also conains the email address from the user as primary key.
To process Commands it is necessary to define so-called ‘Command Handlers’ in Lagom. These determine the Behavior for your PersistentEntity and always start with a clean State. The following listing shows the implementation for the CustomerEntity with its Behavior:

1public class CustomerEntity extends PersistentEntity<CustomerCommand, CustomerEvent, CustomerState> {
2 
3   @Override
4   public Behavior initialBehavior(Optional<CustomerState> snapshotState) {
5 
6      /*
7       * The BehaviorBuilder always starts with a State, which can be initially empty
8       */
9       BehaviorBuilder b = newBehaviorBuilder(
10               snapshotState.orElse(new CustomerState.EMPTY));
11 
12      /*
13       * Command handler for the AddCustomer command.
14       */
15       b.setCommandHandler(CustomerCommand.AddCustomer.class, (cmd, ctx) ->
16               // First we create an event and persist it
17               // {@code entityId() } gives you automatically the 'primary key', in our case the email
18               ctx.thenPersist(new CustomerEvent.AddedCustomerEvent(entityId(), cmd.firstName, cmd.lastName, cmd.birthDate, cmd.comment),
19                       // if this succeeds, we return 'Done'
20                       evt -> ctx.reply(Done.getInstance())));
21 
22      /*
23       * Event handler for the AddedCustomerEvent event, where we update the status for real
24       */
25       b.setEventHandler(CustomerEvent.AddedCustomerEvent.class,
26               evt -> {
27                   return new CustomerState(Optional.of(evt.email), Optional.of(evt.firstName), Optional.of(evt.lastName), Optional.of(evt
28                           .birthDate), evt.comment);
29               });
30 
31      /*
32       * Command handler to query all data of a customer (String representation of our customer)
33       */
34       b.setReadOnlyCommandHandler(CustomerCommand.CustomerInfo.class,
35               (cmd, ctx) -> ctx.reply(state().toString()));
36 
37       return b.build();
38   }
39 
40}

Finally a handler definition in the code listing, a ‘read only command handler’ is being created. You are not allowed to mutate any state through this handler, but it can be used to query the current state of the entity.

The BehaviorBuilder can also contain business logic, for example to mutate state differently when a customer already exists and as such has to be updated instead of created. The AddedCustomerEvent is identical to the AddCustomerCommand except for having the e-mail address, because we’ll need it later on.
Missing until now from the code listings is the CustomerState, which you can see below. The fields are all of type Optional because the initial state for a certain customer is ’empty’.

1public final class CustomerState implements Jsonable {
2 
3   public static final CustomerState EMPTY = new CustomerState(Optional.empty(), Optional.empty, Optional.empty, Optional.empty, Optional.empty);
4 
5   private final Optional<String> email;
6   private final Optional<String> firstName;
7   private final Optional<String> lastName;
8   private final Optional<Date> birthDate;
9   private final Optional<String> comment;
10 
11   @JsonCreator
12   public BlogState(Optional<String> email, Optional<String> firstName, Optional<String> lastName, Optional<Date> birthDate, Optional<String> comment) {
13       this.email = email;
14       this.firstName = firstName;
15       this.lastName = lastName;
16       this.birthDate = birthDate;
17       this.comment = comment;
18   }
19 
20   @JsonIgnore
21   public boolean isEmpty() {
22       return !email.isPresent();
23   }
24}

Read-side with JDBC in Lagom

In a CQRS (Command Query Responsibility Segregation) architecture the manipulation of data is separated from the querying of data. One of the more interesting aspects about this separation is that the read-side can be optimized for querying. Specifically by using denormalized tables on the read-side, grouping data in the most efficient way and by duplicating data where needed. This keeps queries simple and fast.

Additionally this will prevent so-called ORM impedance mismatch; the conceptual and technical difficulties of translating object structures to relational tables, for example translation of inheritance and encapsulation to relational schemas.
As I have shown above Lagom will automatically take care of storage and processing of events in the same way the framework supports storing of data on the read-side inside denormalized tables, shown in Figure 2.

Fig 2: Separated ‘read’ and ‘write’ side in line with CQRS
© Microsoft – CQRS Journey

Within Lagom you can define “ReadSideProcessor”s that can receive and process events and thereby store the data in a different form. The next listing shows an example of a ReadSideProcessor.

1public class CustomerEventProcessor extends ReadSideProcessor<CustomerEvent> {
2 
3   private final JdbcReadSide readSide;
4 
5   @Inject
6   public CustomerEventProcessor(JdbcReadSide readSide) {
7       this.readSide = readSide;
8   }
9 
10   @Override
11   public ReadSideHandler<CustomerEvent> buildHandler() {
12       JdbcReadSide.ReadSideHandlerBuilder<CustomerEvent> builder = readSide.builder("votesoffset");
13 
14       builder.setGlobalPrepare(this::createTable);
15       builder.setEventHandler(CustomerEvent.AddedCustomerEvent.class, this::processCustomerAdded);
16 
17       return builder.build();
18   }
19 
20   private void createTable(Connection connection) throws SQLException {
21       connection.prepareStatement(
22               "CREATE TABLE IF NOT EXISTS customers ( "
23                       + "id MEDIUMINT NOT NULL AUTO_INCREMENT, "
24                       + "email VARCHAR(64) NOT NULL, "
25                       + "firstname VARCHAR(64) NOT NULL, "
26                       + "lastname VARCHAR(64) NOT NULL, "
27                       + "birthdate DATETIME NOT NULL, "
28                       + "comment VARCHAR(256), "
29                       + "dt_created DATETIME DEFAULT CURRENT_TIMESTAMP, "
30                       + " PRIMARY KEY (id))").execute();
31   }
32 
33   private void processCustomerAdded(Connection connection, CustomerEvent.AddedCustomerEvent event) throws SQLException {
34       PreparedStatement statement = connection.prepareStatement(
35               "INSERT INTO customers (email, firstname, lastname, birthdate, comment) VALUES (?, ?, ?, ?, ?)");
36       statement.setString(1, event.email);
37       statement.setString(2, event.firstName);
38       statement.setString(3, event.lastName);
39       statement.setDate(4, event.birthDate);
40       statement.setString(5, event.comment.orElse(""));
41       statement.execute();
42   }
43 
44   @Override
45   public PSequence<AggregateEventTag<CustomerEvent>> aggregateTags() {
46       return TreePVector.singleton(CustomerEvent.CUSTOMER_EVENT_TAG);
47   }
48}

Now the ReadSideProcessor can be registered in the service implementation as follows (showing the full constructor for the sake of completeness):

1@Inject
2public CustomerServiceImpl(PersistentEntityRegistry persistentEntityRegistry, JdbcSession jdbcSession, ReadSide readSide) {
3   this.persistentEntityRegistry = persistentEntityRegistry;
4   this.persistentEntityRegistry.register(CustomerEntity.class);
5   this.jdbcSession = jdbcSession;
6   readSide.register(CustomerEventProcessor.class);
7}

For the Event class a ‘tag’ needs to be defined as shown in the following listing, so Lagom can keep track of which events have been processed. This is important particularly for restarts or crashes, so that the data can be kept consistent between write- and read-side.

1AggregateEventTag<CustomerEvent> CUSTOMER_EVENT_TAG = AggregateEventTag.of(CustomerEvent.class);
2 
3@Override
4default AggregateEventTag<CustomerEvent> aggregateTag() {
5   return CUSTOMER_EVENT_TAG;
6}

Now that the processing of events is implemented and data is stored in denormalized tables, it can be easily queried using SQL queries. For example the next listing shows a simple query for the average age of customers in the system, added to the service implementation.

1@Override
2public ServiceCall<NotUsed, String> getCustomerAverageAge() {
3   return request -> jdbcSession.withConnection(connection -> {
4       ResultSet rsCount = connection.prepareStatement("SELECT COUNT(*) FROM customers").executeQuery();
5       ResultSet rsAverage = connection.prepareStatement("SELECT AVG(TIMESTAMPDIFF(YEAR,birthDate,CURDATE())) FROM customers").executeQuery();
6 
7       if (rsCount.next() && rsAverage.next() && rsCount.getInt(1) > 0) {
8           return String.format("# %s customers resulted in average age; %s", rsCount.getString(1), rsAverage.getString(1));
9       } else {
10           return "No customers yet";
11       }
12   });
13}

Conclusion

CQRS and Event Sourcing are a powerful means to optimize the write- and read-side for a service separately. And while a NoSQL store certainly has its advantages, a relational database is highly suitable for querying over multiple object structures.
I hope to have shown you how Lagom supports this architecture perfectly and supports different solutions for persistence. With the principle of ‘convention over configuration’ developers can focus on implementing business logic instead of typing boilerplate code.

Lagom recently arrived at version 1.2.x and you will sometimes note this is still a young framework in some minor issues. Partly because of this I advise to take some caution and thoroughly evaluate whether Lagom is suitable for your production use-cases. But it certainly is a framework to keep an eye on.