5. Repositories and Event Stores

The FileSystemEventStore stores the events in a file on the file system. It provides good performance and easy configuration. The downside of this event store is that is does not provide transaction support and doesn't cluster very well. The only configuration needed is the location where the event store may store its files and the serializer to use to actually serialize and deserialize the events.

Note that the FileSystemEventStore is not aware of transactions and cannot automatically recover from crashes. Furthermore, it stores a single file for each aggregate, potentially creating too many files for the OS to handle. It is therefore not a suitable implementation for production environments.

Out of disk space

When using the FileSystemEventStore in a test environment (or other environment where many aggregate may be created), you may end up with an "Out of disk space" error, even if there is plenty of room on the disk. The cause is that the filesystem runs out of i-nodes (or an equivalent when not on a Unix filesystem). This typically means that a filesystem holds too many files. To prevent this problem, make sure the output directory of the FileSystemEventStore is cleaned after each test run.

The JpaEventStore stores events in a JPA-compatible data source. Unlike the file system version, the JPAEventStore supports transactions. The JPA Event Store stores events in so called entries. These entries contain the serialized form of an event, as well as some fields where meta-data is stored for fast lookup of these entries. To use the JpaEventStore, you must have the JPA (javax.persistence) annotations on your classpath.

By default, the event store needs you to configure your persistence context (defined in META-INF/persistence.xml file) to contain the classes DomainEventEntry and SnapshotEventEntry (both in the org.axonframework.eventstore.jpa package).

Below is an example configuration of a persistence context configuration:

<persistence xmlns="http://java.sun.com/xml/ns/persistence" version="1.0">
    <persistence-unit name="eventStore" transaction-type="RESOURCE_LOCAL">
        <class>org...eventstore.jpa.DomainEventEntry</class> 
        <class>org...eventstore.jpa.SnapshotEventEntry</class>
    </persistence-unit>
</persistence>

Detecting duplicate key violations in the database

Axon uses Locking to prevent two threads from accessing the same Aggregate. However, if you have multiple JVMs on the same database, this won't help you. In that case, you'd have to rely on the database to detect conflicts. Concurrent access to the event store will result in a Key Constraint Violation, as the table only allows a single Event for an aggregate with any sequence number. Inserting a second event for an existing aggregate with an existing sequence number will result in an error. The JPA EventStore can detect this error and translate it to a ConcurrencyException. However, each database system reports this violation differently. If you register your DataSource with the JpaEventStore, it will try to detect the type of database and figure out which error codes represent a Key Constraint Violation. Alternatively, you may provide a PersistenceExceptionTranslator instance, which can tell if a given exception represents a Key Constraint Violation. If no DataSource or PersistenceExceptionTranslator is provided, exceptions from the database driver are thrown as-is.

public class MyEntityManagerProvider implements EntityManagerProvider {

    private EntityManager entityManager;

    @Override
    public EntityManager getEntityManager() {
        return entityManager;
    }

    @PersistenceContext(unitName = "myPersistenceUnit")
    public void setEntityManager(EntityManager entityManager) {
        this.entityManager = entityManager;
    }                

Customizing the Event storage

By default, the JPA Event Store stores entries in DomainEventEntry and SnapshotEventEntry entities. While this will suffice in many cases, you might encounter a situation where the meta-data provided by these entities is not enough. Or you might want to store events of different aggregate types in different tables.

If that is the case, you may provide your own implementation of EventEntryStore in the JPA Event Store's constructor. You will need to provide implementations of methods that load and store serialized events. Check the API Documentation of the EventEntryStore class for implementation requirements.

If you only want to change the table name or want to add some extra fields to the table, you can also create a class that extends from DefaultEventEntryStore, and override the createDomainEventEntry and/or createSnapshotEventEntryMethod. This method must return a DomainEventEntry and SnapshotEventEntry instance, respectively. By returning your own subclass of these, you can store different events in different tables, or add extra information in separate columns.

Memory consumption warning

Note that persistence providers, such as Hibernate, use a first-level cache on their EntityManager implementation. Typically, this means that all entities used or returned in queries are attached to the EntityManager. They are only cleared when the surrounding transaction is committed or an explicit "clear" in performed inside the transaction. This is especially the case when the Queries are executed in the context of a transaction. To work around this issue, make sure to exclusively query for non-entity objects. You can use JPA's "SELECT new SomeClass(parameters) FROM ..." style queries to work around this issue. Alternatively, call EntityManager.flush() and EntityManager.clear() after fetching a batch of events. Failure to do so might result in OutOfMemoryExceptions when loading large streams of events.

The JDBC event store uses a JDBC Connection to store Events in a JDBC compatible data storage. Typically, these are relational databases. Theoretically, anything that has a JDBC driver could be used to back the JDBC Event Store.

Similar to the JPA Event Store, the JDBC Event Store stores Events in entries. By default, each Event is stored in a single Entry, which corresponds with a row in a table. One table is used for Events and another for the Snapshots.

The JdbcEventStore can be configured with an EventEntryStore and a Serializer. The EventEntryStore defines how Events are appended to the event store. The serializer is used to convert the payload and meta data of an event into an array of bytes, ready for storage. In most cases, the DefaultEventEntryStore will suffice. t can be configured to accommodate all sort of different scenarios.

The DefaultEventEntryStore uses a ConnectionProvider to obtain connections. Typically, these connections can be obtained directly from a DataSource. However, Axon will bind these connections the a Unit of Work, so that a single connection in a Unit of Work. This ensures that a single transaction is used to store all events, even when multiple Unit of Work are nested in the same thread.

JDBC Event Store and Spring

Spring users are recommended to use the namespace support to define a JDBC Event Store: <axon:jdbc-event-store .../> This will ensure that connections are bound to a transaction using the Platform Transaction Manager, if available. If you don't use namespace support, or define your own ConnectionProvider, you can use the SpringDataSourceConnectionProvider to attach a connection from a DataSource to an existing transaction. It is also recommended to wrap the ConnectionProvider in a UnitOfWorkAwareConnectionProviderWrapper, to ensure a single connection is used during the course of a single Unit of Work.

Most database speak more or less the same language. However, there many so-called SQL Dialects. While the JDBC Event Store speaks a language all database should be able to understand, it is possible that specific database vendors provide better performing alternatives to generic SQL commands. To accommodate those, the DefaultEventEntryStore works with a EventSqlSchema. The EventSqlSchema is an interface that prescribes a number of operations the EventEntryStore does on the underlying database. The EventSqlSchema is responsible for creating the correct PreparedStatements for those. When you need to change a query that is executed against the database, it will usually suffice to override a single method in the GenericEventSqlSchema.

Timestamps and time zones

By default, Axon stores time stamps in the system timezone. However, many regions use daylight savings time, causing them to effectively change timezone throughout the year. This could cause events to be returned in a different order than how they were originally stored. It is recommended to store timestamps in the UTC timezone, or use the millis-since-epoch format. To force te JDBC Event Store to store dates in the UTC timezone, either configure Joda to generate all dates in UTC timezone, or tell the JDBC Event Store to convert all timestamps to UTC. This can be done by setting <axon:jdbc-event-store ... force-utc-timestamp="true" ... />, or by calling setForceUtc(true); on the GenericEventSqlSchema.

MongoDB is a document based NoSQL store. It scalability characteristics make it suitable for use as an Event Store. Axon provides the MongoEventStore, which uses MongoDB as backing database. It is contained in the Axon Mongo module (Maven artifactId axon-mongo).

Events are stored in two separate collections: one for the actual event streams and one for the snapshots.

By default, the MongoEventStore stores each event in a separate document. It is, however, possible to change the StorageStrategy used. The alternative provided by Axon is the DocumentPerCommitStorageStrategy, which creates a single document for all Events that have been stored in a single commit (i.e. in the same DomainEventStream).

Storing an entire commit in a single document has the advantage that a commit is stored atomically. Furthermore, it requires only a single roundtrip for any number of events. A disadvantage is that it becomes harder to query events manually or through the EventStoreManagement methods. When refactoring the domain model, for example, it is harder to "transfer" events from one aggregate to another if they are included in a "commit document".

The MongoDB doesn't take a lot of configuration. All it needs is a reference to the collections to store the Events in, and you're set to go. For production environments, you may want to double check the indexes on your collections.

If you have specific requirements for an event store, it is quite easy to implement one using different underlying data sources. Reading and appending events is done using a DomainEventStream, which is quite similar to iterator implementations.

Instead of eagerly deserializing Events, consider using the SerializedDomainEventMessage, which will postpone deserialization of Meta Data and Payload until it is actually used by a handler.

	Tip
	The SimpleDomainEventStream class will make the contents of a sequence ( List or array) of EventMessage instances accessible as event stream.

Event Stores need a way to serialize the Event to prepare it for storage. By default, Axon uses the XStreamSerializer, which uses XStream to serialize Events into XML. XStream is reasonably fast and is more flexible than Java Serialization. Furthermore, the result of XStream serialization is human readable. Quite useful for logging and debugging purposes.

The XStreamSerializer can be configured. You can define aliases it should use for certain packages, classes or even fields. Besides being a nice way to shorten potentially long names, aliases can also be used when class definitions of event change. For more information about aliases, visit the XStream website.

Alternatively, Axon also provides the JacksonSerializer, which uses Jackson to serialize Events into JSON. While it produces more a compact serialized form, it does require that classes stick to the conventions (or configuration) required by Jackson.

	Spring XML Configuration and Serializer Customization
	Configuring the serializer using Java code (or other JVM languages) is easy. However, configuring it in a Spring XML Application Context is not so trivial, due to its limitations to invoke methods. One of the options is to create a FactoryBean that creates an instance of an XStreamSerializer and configures it in code. Check the Spring Reference for more information.

You may also implement your own Serializer, simply by creating a class that implements Serializer, and configuring the Event Store to use that implementation instead of the default.

To explain how to write an upcaster for Axon we'll walk through a small example, describing the details of writing an upcaster as we go along.

Let's assume that there is an Event Store containing many AdministrativeDetailsUpdated events. New requirements have let to the introduction of two new events: AddressUpdatedEvent and InsurancePolicyUpdatedEvent. Previously though, all information in these two events was contained in the old AdministrativeDetailsUpdatedEvent, which is now deprecated. To nicely handle this situation we'll write an upcaster to transform the AdministrativeDetailsUpdatedEvent into an AddressUpdatedEvent and an InsurancePolicyUpdatedEvent.

Here is the code for an upcaster:

import org.dom4j.Document;
                
public class AdministrativeDetailsUpdatedUpcaster implements Upcaster<Document> { 

    @Override
    public boolean canUpcast(SerializedType serializedType) { 
        return serializedType.getName().equals("org.example.AdministrativeDetailsUpdated") &&
               "0".equals(serializedType.getRevision());
    }

    @Override
    public Class<Document> expectedRepresentationType() { 
        return Document.class;
    }

    @Override
    public List<SerializedObject<Document>> upcast(SerializedObject<Document> intermediateRepresentation,
                                                   List<SerializedType> expectedTypes, UpcastingContext context) {
        Document administrativeDetailsUpdatedEvent = intermediateRepresentation.getData();

        Document addressUpdatedEvent =
                new DOMDocument(new DOMElement("org.example.InsurancePolicyUpdatedEvent"));
        addressUpdatedEvent.getRootElement()
                .add(administrativeDetailsUpdatedEvent.getRootElement().element("address").createCopy());

        Document insurancePolicyUpdatedEvent =
                new DOMDocument(new DOMElement("org.example.InsurancePolicyUpdatedEvent").createCopy());
        insurancePolicyUpdatedEvent.getRootElement()
                .add(administrativeDetailsUpdatedEvent.getRootElement().element("policy").createCopy());

        List<SerializedObject<?>> upcastedEvents = new ArrayList<SerializedObject<?>>();
        upcastedEvents.add(new SimpleSerializedObject<Document>(
                addressUpdatedEvent, Document.class, expectedTypes.get(0)));
        upcastedEvents.add(new SimpleSerializedObject<Document>(
                insurancePolicyUpdatedEvent, Document.class, expectedTypes.get(1)));
        return upcastedEvents;
    }

    @Override
    public List<SerializedType> upcast(SerializedType serializedType) { 
        SerializedType addressUpdatedEventType = 
                new SimpleSerializedType("org.example.AddressUpdatedEvent", "1");
        SerializedType insurancePolicyUpdatedEventType = 
                new SimpleSerializedType("org.example.InsurancePolicyUpdatedEvent", "1");
        return Arrays.asList(addressUpdatedEventType, insurancePolicyUpdatedEventType);
    }
}

Conditional upcasting

In some occasions, it is necessary to upcast an event to one of multiple potential types, based on the contents of the event itself. This is the case where one historical event has been split into several new events, each one for a specific case. The Upcaster interface doesn't provide the event itself in the upcast(SerializedType) method. Alternatively, you may implement the ExtendedUpcaster interface. Unlike the Upcaster interface, this interface declares an upcast(SerializedType, SerializedObject) method. This method serves as a replacement for the upcast(SerializedType) method, for UpcasterChain implementations that support ExtendedUpcaster. The UpcasterChain implementations provided by Axon will always support this. Upcasters implementing ExtendedUpcaster may throw an UnsupportedOperationException in the upcast(SerializedType) method.

The Upcaster Chain is responsible for upcasting events by chaining the output of one upcaster to the next. It comes in the following two flavours:

The LazyUpcasterChain is a safe choice if your upcasters are stateless or do not depend on other upcasters. Always consider using the LazyUpcasterChain since it can provide a great performance benefit over the SimpleUpcasterChain. If you want guaranteed upcasting in a strict order, use the SimpleUpcasterChain.

An upcaster works on a given content type (e.g. dom4j Document). To provide extra flexibility between upcasters, content types between chained upcasters may vary. Axon will try to convert between the content types automatically by using ContentTypeConverters. It will search for the shortest path from type x to type y, perform the conversion and pass the converted value into the requested upcaster. For performance reasons, conversion will only be performed if the canUpcast method on the receiving upcaster yields true.

The ContentTypeConverters may depend on the type of serializer used. Attempting to convert a byte[] to a dom4j Document will not make any sense unless a Serializer was used that writes an event as XML. To make sure the UpcasterChain has access to the serializer-specific ContentTypeConverters, you can pass a reference to the serializer to the constructor of the UpcasterChain.

	Tip
	To achieve the best performance, ensure that all upcasters in the same chain (where one's output is another's input) work on the same content type.

If the content type conversion that you need is not provided by Axon you can always write one yourself using the ContentTypeConverter interface.

The XStreamSerializer supports Dom4J as well as XOM as XML document representations. The JacksonSerializer supports Jackson's JsonNode.

Snapshot creation can be triggered by a number of factors, for example the number of events created since the last snapshot, the time to initialize an aggregate exceeds a certain threshold, time-based, etc. Currently, Axon provides a mechanism that allows you to trigger snapshots based on an event count threshold.

The EventCountSnapshotterTrigger provides the mechanism to trigger snapshot creation when the number of events needed to load an aggregate exceeds a certain threshold. If the number of events needed to load an aggregate exceeds a certain configurable threshold, the trigger tells a Snapshotter to create a snapshot for the aggregate.

The snapshot trigger is configured on an Event Sourcing Repository and has a number of properties that allow you to tweak triggering:

A Snapshotter is responsible for the actual creation of a snapshot. Typically, snapshotting is a process that should disturb the operational processes as little as possible. Therefore, it is recommended to run the snapshotter in a different thread. The Snapshotter interface declares a single method: scheduleSnapshot(), which takes the aggregate's type and identifier as parameters.

Axon provides the AggregateSnapshotter, which creates and stores AggregateSnapshot instances. This is a special type of snapshot, since it contains the actual aggregate instance within it. The repositories provided by Axon are aware of this type of snapshot, and will extract the aggregate from it, instead of instantiating a new one. All events loaded after the snapshot events are streamed to the extracted aggregate instance.

	Note
	Do make sure that the Serializer instance you use (which defaults to the XStreamSerializer) is capable of serializing your aggregate. The XStreamSerializer requires you to use either a Hotspot JVM, or your aggregate must either have an accessible default constructor or implement the Serializable interface.

The AbstractSnapshotter provides a basic set of properties that allow you to tweak the way snapshots are created:

The AggregateSnapshotter provides on more property:

Note

If you use an executor that executes snapshot creation in another thread, make sure you configure the correct transaction management for your underlying event store, if necessary. Spring users can use the SpringAggregateSnapshotter, which allows you to configure a PlatformTransactionManager. The SpringAggregateSnapshotter will autowire all aggregate factories (either directly, or via the Repository), if a list is not explicitly configured.

All Axon-provided Event Store implementations are capable of storing snapshot events. They provide a special method that allows a DomainEventMessage to be stored as a snapshot event. You have to initialize the snapshot event completely, including the aggregate identifier and the sequence number. There is a special constructor on the GenericDomainEventMessage for this purpose. The sequence number must be equal to the sequence number of the last event that was included in the state that the snapshot represents. In most cases, you can use the getVersion() on the AggregateRoot (which each event sourced aggregate implements) to obtain the sequence number to use in the snapshot event.

When a snapshot is stored in the Event Store, it will automatically use that snapshot to summarize all prior events and return it in their place. All event store implementations allow for concurrent creation of snapshots. This means they allow snapshots to be stored while another process is adding Events for the same aggregate. This allows the snapshotting process to run as a separate process altogether.

	Note
	Normally, you can archive all events once they are part of a snapshot event. Snapshotted events will never be read in again by the event store in regular operational scenario's. However, if you want to be able to reconstruct aggregate state prior to the moment the snapshot was created, you must keep the events up to that date.

Axon provides a special type of snapshot event: the AggregateSnapshot, which stores an entire aggregate as a snapshot. The motivation is simple: your aggregate should only contain the state relevant to take business decisions. This is exactly the information you want captured in a snapshot. All Event Sourcing Repositories provided by Axon recognize the AggregateSnapshot, and will extract the aggregate from it. Beware that using this snapshot event requires that the event serialization mechanism needs to be able to serialize the aggregate.

A snapshot event is an event like any other. That means a snapshot event is handled just like any other domain event. When using annotations to demarcate event handlers (@EventHandler), you can annotate a method that initializes full aggregate state based on a snapshot event. The code sample below shows how snapshot events are treated like any other domain event within the aggregate.

public class MyAggregate extends AbstractAnnotatedAggregateRoot {

    // ... code omitted for brevity

    @EventHandler
    protected void handleSomeStateChangeEvent(MyDomainEvent event) {
        // ...
    }

    @EventHandler
    protected void applySnapshot(MySnapshotEvent event) {
        // the snapshot event should contain all relevant state
        this.someState = event.someState;
        this.otherState = event.otherState;
    }
}                

There is one type of snapshot event that is treated differently: the AggregateSnapshot. This type of snapshot event contains the actual aggregate. The aggregate factory recognizes this type of event and extracts the aggregate from the snapshot. Then, all other events are re-applied to the extracted snapshot. That means aggregates never need to be able to deal with AggregateSnapshot instances themselves.

Once a snapshot event is written, it prevents older events and snapshot events from being read. Domain Events are still used in case a snapshot event becomes obsolete due to changes in the structure of an aggregate. The older snapshot events are hardly ever needed. SnapshotEventStore implementation may choose to keep only a limited amount of snapshots (e.g. only one) for each aggregate.

The JpaEventStore allows you to configure the amount of snapshots to keep per aggregate. It defaults to 1, meaning that only the latest snapshot event is kept for each aggregate. Use setMaxSnapshotsArchived(int) to change this setting. Use a negative integer to prevent pruning altogether.