Event Sourcing with JVM Languages

Rahul Somasunderam

About Me

Wanted Event Sourcing in 2016
Didn’t like anything I saw
Built my own thing - Grooves
I solve problems in Health Care

Where I work

Transcend Insights® empowers population health management with The Platform Above—technology that rises above the data silos and connects care teams with the insights they need to succeed in a value-based care world.

About You

What is Event Sourcing

Treat your database like you treat your application logs

What does that mean

Create

Read

Update

Delete

Bad Example 1

Mark and Jane share a bank account. The account has a balance of $75. Mark goes to an ATM and withdraws $50. At the same time Jane attempts to withdraw $50.

This is what our code looks like

@Transactional
boolean withdraw(String accountNumber, double amount) {
  double balance = accountService.getBalance(accountNumber);
  if (balance > amount) {
    balance -= account;
    accountService.setBalance(accountNumber, balance);
    return true;
  } else {
    return false;
  }
}

What banks typically do

By Deutsche Bundespost (Scan by User:Mattes) - Scan (700dpi, Millionen Farben), Public Domain Wikimedia

But that was Banking

If you can look at your logs and debug your application, you are already doing that.

Banks defined their business model this way hundreds of years ago.

What about my domain?

/* Aggregate */
class Account {
  String accountNumber;
}

/* Events */
abstract class Transaction {
  Account account;
}
class AtmWithdrawal
    extends Transaction {
  String location;
  double amount;
}
class AtmDeposit
    extends Transaction {
  String location;
  double amount;
}

/* Snapshot */
class AccountSummary {
  double balance;
}

What about my domain?

/* Aggregate */
class Patient {
  String identifier;
  String system;
}

/* Events */
abstract class PatientEvent {
  Patient patient;
}
class MedicationPrescribed
    extends PatientEvent {
  String code;
  int quantity;
}
class ProcedurePerformed
    extends PatientEvent {
  String code;
}

/* Snapshot */
class PatientSummary {
  List<String> medications;
  List<String> procedures;
}

Computing Snapshots

\$S_N = f(S_0 , E_1..E_N)\$

\$S_N\$

Snapshot at Version N

\$S_0\$

empty snapshot

\$E_1..E_N\$

events from position 1 through N

\$f\$

query

But it’s not that simple

Incremental Computation

This is what we would like to do

\$S_N = f(S_K , E_(K+1)..E_N)\$

In Mathematics, it’s called the distributive property

\$a+b+c = (a+b) +c\$

We’re not done yet

There are special events

Revert

Revert Again

Merge/Deprecate

Reverting a Merge

Bad Example 2

Employee

ID	EMP_NAME	DEPT_ID
1	Mr Spock	1
2	Scotty	2
3	Kirk	3
4	Janeway	3
5	La Forge	2

Department

ID	DEPT_NAME
1	Science
2	Engineering
3	Command

Joins in Event Sourcing

Disjoins too

Grooves Domain Objects

Aggregates

public class Patient implements AggregateType<Long> {
    private Long id;
    private String uniqueId;
}

Events

public abstract class PatientEvent implements
        BaseEvent<Long, Patient, Long, PatientEvent> {
    private Patient aggregate;
    private Long id;
    private String createdBy;
    private RevertEvent<Long, Patient, Long, PatientEvent> revertedBy;
    private Date timestamp;
    private Long position;

    @Override
    @NotNull
    public Observable<Patient> getAggregateObservable() {
        return aggregate != null ? just(aggregate) : empty();
    }
}

This is a base class for all events that we’ll be applying on Patient. All these properties are required to make Grooves happy.

createdBy is where we’ll put in some information on the authorization.

revertedBy is going to be transient from a database perspective. That will allow us to build an audit ui that tells us why a certain event was reverted. It will point to a Revert Event.

timestamp and position are going to help us work with a timeline. Sometimes we need to look at the time of an event. Sometimes we just care about the position of an event.

The position is interesting, in that it could be global, or it could be specific to an aggregate.

If you make it global, you can answer questions about the state of the whole system. That works a bit like SVN used to work for versioning code.

If you make it local, then it works more locally, but it can then be used for optimistic locking. You can try to do optimistic locking with global positioning, but the effects on scalability might not be good.

You’ll notice that the type params are already getting long, and they will continue to get longer as we work on queries. So I’ll skip type parameters occasionally here. If you end up using grooves, the documentation will guide you on them.

Real Events

public class PatientCreated extends PatientEvent {
    private String name;
}

Special Events

public class PatientEventReverted
        extends PatientEvent
        implements RevertEvent<Long, Patient, Long, PatientEvent> {
    private Long revertedEventId;
}

Snapshots

public class PatientAccount
        implements JavaSnapshot<Long, Patient, Long, Long, PatientEvent>,
        Serializable {
    private Long id;
    private Patient aggregate;
    private Patient deprecatedBy;
    private List<Patient> deprecates = new ArrayList<>();
    private Long lastEventPosition;
    private Date lastEventTimestamp;

    private String name;
    private BigDecimal balance = new BigDecimal(0);
    private BigDecimal moneyMade = new BigDecimal(0);

    public Observable<Patient> getAggregateObservable() {
        return just(aggregate);
    }

    public Observable<Patient> getDeprecatedByObservable() {
        return just(deprecatedBy);
    }

    public Observable<Patient> getDeprecatesObservable() {
        return from(deprecates);
    }
}

In this case, we’re looking at a snapshot of a patient that will tell us how much is owed by the patient, and how much money has been made. Pretty simple: The highlighted portions show us those properties. The rest is to make Grooves happy.

deprecatedBy tells you if the aggregate that this snapshot represents has been deprecated by another aggregate. Think of the Freddie Mercury and Farokh Bulsara example.

deprecates tells you of all the aggregates that were deprecated by the aggregate of this snapshot. That helps you look up the deprecated ones to see if you need to include any event in your computation.

If you’re just computing bank balances, it’s easy. If you’re doing something complex like looking at multiple medications and deciding if something needs to be done for a patient, you can’t take shortcuts.

Grooves Queries

Satisfying grooves

public class PatientAccountQuery<...> extends QuerySupport<...> {
    ...
}

Could also be VersionedQuerySupport or TemporalQuerySupport

Fetching snapshots

    @Override
    public PatientAccount createEmptySnapshot() {
        return new PatientAccount();
    }

    @Override
    public Observable<PatientAccount> getSnapshot(
            long maxPosition, Patient aggregate) {
        ...
    }

    @Override
    public Observable<PatientAccount> getSnapshot(
            Date maxTimestamp, Patient aggregate) {
        ...
    }

Fetching events

    @Override
    public Observable<PatientEvent> getUncomputedEvents(
            Patient aggregate, PatientAccount lastSnapshot, long version) {

    }

    @Override
    public Observable<PatientEvent> getUncomputedEvents(
            Patient aggregate, PatientAccount lastSnapshot, Date snapshotTime) {

    }

Handling errors

    @Override
    default Observable<EventApplyOutcome> onException(
            Exception e, PatientAccount snapshot, PatientEvent event) {
        getLog().error("Error computing snapshot", e);
        return just(CONTINUE);
    }

Sometimes your code can throw exceptions. If it does, we need a place to catch these exceptions and decide what to do. We might just take the simplest path and log the exception and abort the computation for this snapshot.

Or, you could collect all the exceptions, and then present them to the user in a simpler form. This way, you can tell the user that you tried to do your best, but that wasn’t good enough.

So, though you can present them with a snapshot, they should take it with a grain of salt.

You might not want to do that for a financial transaction, but if you’re doing something for health care, this is much better than an error message. The doctor can look your condition, and the warning, and then ask the patient, "Hey, you’ve got high blood pressure, but I don’t see your blood pressure medication. Are you taking one? Let me fix the record here…"

Processing events

For languages like Java, Groovy

    public Observable<EventApplyOutcome> applyPatientCreated(
            PatientCreated event, PatientAccount snapshot) {
        if (snapshot.getAggregate() == event.getAggregate()) {
            snapshot.setName(event.getName());
        }
        return just(CONTINUE);
    }

    ...

Processing events

For languages with case classes

    override fun applyEvent(
        event: PatientEvent.Applicable, snapshot: PatientAccount) =
            when (event) {
                is PatientEvent.Applicable.Created -> {
                    // Your logic here
                    just(CONTINUE)
                }
                is PatientEvent.Applicable.ProcedurePerformed -> {
                    // Your logic here
                    just(CONTINUE)
                }
                is PatientEvent.Applicable.PaymentMade -> {
                    // Your logic here
                    just(CONTINUE)
                }
            }

Events

For languages with case classes

sealed class PatientEvent : BaseEvent<..> {
    // Properties and methods from equivalent java class

    sealed class Applicable : PatientEvent() {
        data class Created(val name: String) : Applicable()
        data class ProcedurePerformed(
                val code: String, val cost: Double) : Applicable()
        ..
    }

    data class Reverted(override val revertedEventId: String) :
            PatientEvent(), RevertEvent<..>
}

Completeness for Java

…and Groovy.

@Aggregate public class Patient {...}
public abstract class PatientEvent {}

@Event(Patient.class)
public class ProcedurePerformed extends PatientEvent {}
@Event(Patient.class)
public class PaymentMade extends PatientEvent {}

@Query(aggregate=Patient.class, snapshot=PatientAccount.class)
public class PatientAccountQuery {
    ...
    Observable<EventApplyOutcome> applyProcedurePerformed() {...}
    Observable<EventApplyOutcome> applyPaymentMade() {...}
}

Usage

Patient patient = ...
Observable<PatientAccount> account;

// By Date
account = patientAccountQuery.computeSnapshot(patient, new Date());

// By version
account = patientAccountQuery.computeSnapshot(patient, 7L);

Gotchas

How many Tour de France General Classification Tour victories did this guy have on 2007-10-01?

EPA/JASPER JUINEN, 22 July 2004. Sourced from vosizneias

Lance Armstrong

1992	Joins Motorola
1993	DNF
1994	DNF
1995	36
1996	DNF
1996-10	Diagnosed with Cancer
1997-02	Declared cancer free
1998	Joins US Postal
1999
2000
2001

2002
2003
2004
2005
2005	Retires
2009	Returns from retirement
2009	3
2010	23
2010	Retires
2012-10	Stripped of all wins from 1998 through 2010

1. Based on what we knew in 2007, How many TdF GC Tour victories did he have on 2007-10-01?

2. Based on what we know in 2017, How many TdF GC Tour victories did he have on 2007-10-01?

Ask the right question

Advice on handling data

Be very careful about what your events look like.
Don’t worry much about what your snapshots look like.

Communication is key

95% of the cost of event sourcing projects is explaining event sourcing
— Scott Bellware (@sbellware) June 11, 2017

Non Event Sources of data

When the intent of the user is not clear from the datasource.

The Story of Goldilocks and The Three Bears, Award Publications LTD

Thank you

Slidedeck

bit.ly/2017-es

Grooves

github.com/rahulsom/grooves

Sun 15:00

CON7610

Microservices Data Patterns: CQRS and ES

Mon 16:30

CON2526

Reactive Stream Processing with Swarm and Kafka

Mon 17:30

CON7474

ES, Distributed Systems, and CQRS with Java EE

Tue 13:15

CON4083

Async by Default, Synchronous When Necessary

Wed 14:45

CON4277

Three µS Patterns to Tear Down Your Monoliths