Event Sourcing Template

Overview

Event Sourcing is Talon's High Availability model that provides the best performance for latency-sensitive applications. With Event Sourcing, Talon replicates inbound messages rather than state changes. Each instance processes the same sequence of messages deterministically to maintain identical state and produce identical outputs.

How Event Sourcing Works

With Event Sourcing:

• Inbound messages are replicated - Messages are replicated to backup instances in parallel with processing

• State is application-managed - Your state is stored in POJOs, opaque to the Talon runtime

• Deterministic processing ensures consistency - All instances execute identical code on identical inputs

• No state modeling required - State can be any Java objects (ADM modeling optional)

• Lower replication overhead - Messages replicated in serialized form without re-serialization

Key Features

• Application-Controlled State: Complete control over state structure and management

• Opaque State: Runtime doesn't inspect or manage your state

• Low Latency: Messages replicated in parallel with handler execution

• Efficient Replication: Received messages already serialized, no re-serialization cost

• Full Flexibility: Use any Java objects for state, no ADM constraints

• Deterministic Recovery: State reconstructed through message replay

When to Use Event Sourcing

Event Sourcing is ideal when:

• Ultra-low latency is critical (sub-millisecond requirements)

• You need full control over state management

• State logic is complex or involves custom algorithms

• Your state doesn't fit ADM modeling constraints

• You want to use standard Java POJOs

• Replication overhead must be minimized

Comparison with State Replication

Aspect	Event Sourcing	State Replication
State Management	Application manages (opaque POJOs)	Runtime manages (transparent)
State Modeling	No modeling required	ADM XML required
Replication	Inbound messages replicated	State deltas replicated
Latency	Lower (parallel replication)	Slightly higher (state tracking)
Complexity	More complex (determinism requirements)	Simpler application code
Recovery	State reconstructed by message replay	State reconstructed from log
Discipline Required	High (must avoid divergence)	Low (runtime enforces consistency)

See Consensus Models for detailed comparison.

---

Building an Event Sourcing Microservice

Creating an Event Sourcing microservice involves five main steps:

1. Model your application messages using ADM

2. Annotate your main class for Event Sourcing

3. Manage microservice state in POJOs

4. Write deterministic message handlers

5. Configure storage (clustering and persistence)

Step 1: Model Application Messages

Define your microservice's messages using the ADM modeling language in your model.xml file. Unlike State Replication, you don't need to model your state in ADM - only your messages.

Example Message Model:

<model xmlns="http://www.neeveresearch.com/schema/x-ddl"
       namespace="com.example.orderprocessor.messages"
       defaultFactoryId="1">

  <messages>
    <!-- Inbound message -->
    <message name="NewOrderRequest" id="1">
      <field name="orderId" type="String"/>
      <field name="symbol" type="String"/>
      <field name="quantity" type="Integer"/>
      <field name="price" type="BigDecimal"/>
    </message>

    <!-- Outbound message -->
    <message name="OrderConfirmation" id="2">
      <field name="orderId" type="String"/>
      <field name="status" type="String"/>
      <field name="timestamp" type="Long"/>
    </message>

    <message name="CancelOrderRequest" id="3">
      <field name="orderId" type="String"/>
    </message>
  </messages>
</model>

See The Modeling Language for complete modeling guide.

Step 2: Annotate Main Class for Event Sourcing

Use the @AppHAPolicy annotation to declare that your microservice uses Event Sourcing:

package com.example.orderprocessor;

import com.neeve.aep.AepEngine;
import com.neeve.aep.annotations.AppHAPolicy;

@AppHAPolicy(value = AepEngine.HAPolicy.EventSourcing)
public class OrderProcessorApp {
    // ... application code ...
}

Step 3: Manage Application State

With Event Sourcing, your state is completely private to your application. You can use any Java objects - POJOs, collections, or even third-party data structures:

public class OrderProcessorApp {
    // Private POJO state - opaque to Talon
    private class OrderBook {
        private long totalOrders = 0;
        private BigDecimal totalValue = BigDecimal.ZERO;
        private final Map<String, Order> orders = new HashMap<>();

        public void addOrder(Order order) {
            orders.put(order.getOrderId(), order);
            totalOrders++;
            totalValue = totalValue.add(
                order.getPrice().multiply(BigDecimal.valueOf(order.getQuantity()))
            );
        }

        public Order getOrder(String orderId) {
            return orders.get(orderId);
        }

        public long getTotalOrders() {
            return totalOrders;
        }
    }

    // Simple POJO for order state
    private static class Order {
        private String orderId;
        private String symbol;
        private int quantity;
        private BigDecimal price;
        private String status;

        // Constructor, getters, setters...
    }

    // Application state instance
    private final OrderBook orderBook = new OrderBook();
}

Important Points:

• State is completely private to your application

• No ADM modeling required for state

• Runtime never inspects or manages your state

• State must be reconstructed deterministically on recovery through message replay

Step 4: Write Deterministic Message Handlers

Message handlers must be deterministic - they must produce identical state and outputs given identical inputs. Handlers receive only the inbound message (not state, since state is private).

import com.neeve.aep.AepMessageSender;
import com.neeve.aep.AepEngine;
import com.neeve.aep.annotations.EventHandler;
import com.neeve.aep.annotations.AppInjectionPoint;

public class OrderProcessorApp {
    private AepMessageSender messageSender;
    private AepEngine engine;
    private final OrderBook orderBook = new OrderBook();

    // Inject message sender for outbound messages
    @AppInjectionPoint
    final public void setMessageSender(AepMessageSender messageSender) {
        this.messageSender = messageSender;
    }

    // Inject engine for accessing engine time
    @AppInjectionPoint
    final public void initialize(AepEngine engine) {
        this.engine = engine;
        // Register message factory
        engine.registerFactory(new com.example.orderprocessor.messages.MessageFactory());
    }

    @EventHandler
    final public void onNewOrder(NewOrderRequest request) {
        // Create order from request - deterministic
        Order order = new Order();
        order.setOrderId(request.getOrderId());
        order.setSymbol(request.getSymbol());
        order.setQuantity(request.getQuantity());
        order.setPrice(request.getPrice());
        order.setStatus("ACCEPTED");

        // Update private POJO state
        orderBook.addOrder(order);

        // Send outbound message
        OrderConfirmation confirmation = OrderConfirmation.create();
        confirmation.setOrderId(order.getOrderId());
        confirmation.setStatus("ACCEPTED");
        // IMPORTANT: Use engine.getEngineTime() for deterministic time
        confirmation.setTimestamp(engine.getEngineTime());
        messageSender.sendMessage("confirmations", confirmation);
    }

    @EventHandler
    final public void onCancelOrder(CancelOrderRequest request) {
        Order order = orderBook.getOrder(request.getOrderId());
        if (order != null && "ACCEPTED".equals(order.getStatus())) {
            // Update state deterministically
            order.setStatus("CANCELLED");

            // Send confirmation
            OrderConfirmation confirmation = OrderConfirmation.create();
            confirmation.setOrderId(order.getOrderId());
            confirmation.setStatus("CANCELLED");
            confirmation.setTimestamp(engine.getEngineTime());
            messageSender.sendMessage("confirmations", confirmation);
        }
    }
}

Critical: Preventing Divergence

Event Sourcing requires special discipline to avoid state divergence between primary and backup:

Never use non-deterministic operations in handlers:

• System.currentTimeMillis() - Use engine.getEngineTime() instead

• System.nanoTime() - Not replicated, will cause divergence

• Random numbers - Will differ on backup

• External I/O or database calls - Results not replicated

• Environment variables - May differ between instances

• Clock-dependent logic - Clocks may be skewed

Safe Practices:

• Use engine.getEngineTime() - Provides deterministic timestamp from message

• Process only message data - All decisions based on message content and existing state

• Use message injection - Inject results from external operations as new messages

• Use environment replication - Tunnel environment data into replication stream (see below)

Step 5: Configure Storage

Configure storage in your DDL to enable clustering and persistence.

Register Message Factories

Only message factories need to be registered (not state factories, since state is opaque):

DDL Configuration:

<app name="order-processor" mainClass="com.example.orderprocessor.OrderProcessorApp">
  <messaging>
    <factories>
      <factory name="com.example.orderprocessor.messages.MessageFactory"/>
    </factories>
    <buses>
      <bus name="orders-bus">
        <channels>
          <channel name="orders" join="true"/>
          <channel name="confirmations" join="false"/>
        </channels>
      </bus>
    </buses>
  </messaging>

  <storage enabled="true">
    <clustering enabled="true">
      <storeName>order-processor-store</storeName>
      <localPort>10000</localPort>
      <discoveryDescriptor>nvds://localhost:4090</discoveryDescriptor>
    </clustering>
    <persistence enabled="true">
      <flushOnCommit>false</flushOnCommit>
      <storeRoot>rdat</storeRoot>
    </persistence>
  </storage>
</app>

Programmatic Registration Alternative:

@AppInjectionPoint
public void initialize(AepEngine engine) {
    // Register message factory for deserialization
    engine.registerFactory(new com.example.orderprocessor.messages.MessageFactory());
}

Enable Clustering

Clustering allows multiple instances to discover each other and form an HA cluster:

<clustering enabled="true">
  <storeName>order-processor-store</storeName>
  <localPort>10000</localPort>
  <localIfAddr>192.168.1.100</localIfAddr>
  <linkParams>SO_REUSEADDR=true,TCP_NODELAY=true</linkParams>
  <discoveryDescriptor>nvds://localhost:4090</discoveryDescriptor>
  <initWaitTime>5000</initWaitTime>
  <failOnMultiplePrimaries>true</failOnMultiplePrimaries>
  <memberElectionPriority>100</memberElectionPriority>
</clustering>

Key clustering concepts:

• Instances with same storeName form a cluster

• One instance elected as primary via leadership election

• Primary establishes messaging and invokes handlers

• Backups receive replicated messages and process them identically

Enable Persistence

Persistence logs the message stream to disk for cold start recovery:

<persistence enabled="true">
  <flushOnCommit>false</flushOnCommit>
  <autoFlushSize>8192</autoFlushSize>
  <storeRoot>rdat</storeRoot>
  <compaction>
    <compactOnStart>false</compactOnStart>
    <compactionThreshold>1024</compactionThreshold>
  </compaction>
</persistence>

Clustering Requires Persistence: When clustering is enabled, persistence must also be enabled. The transaction log is used to initialize new cluster members that connect to the primary.

Cold Start Recovery: With Event Sourcing, recovery from a cold start (no backup running) requires replaying the entire inbound message stream from disk. For long-running applications with high message volumes, transaction log compaction and checkpointing strategies are critical.

See Storage Configuration for complete configuration reference.

---

Advanced Techniques

Using Engine Time

The AepEngine.getEngineTime() method provides deterministic timestamps for Event Sourcing applications:

private volatile AepEngine engine;

@AppInjectionPoint
public void initialize(AepEngine engine) {
    this.engine = engine;
}

@EventHandler
public void onMessage(TradeUpdate message) {
    OrderEvent event = OrderEvent.create();

    // Engine time is identical on primary and backup
    event.setTimeReceivedAsTimestamp(engine.getEngineTime());

    messageSender.sendMessage("order-events", event);
}

How It Works:

• When called from a handler in Event Sourcing mode, returns timestamp from message event

• Timestamp is captured just before handler dispatch

• Same timestamp replicated to backup, ensuring deterministic processing

• Outside handlers or in State Replication mode, returns System.currentTimeMillis()

Message Injection

Message injection allows you to inject messages into the processing stream for fault-tolerant execution. This is useful for:

• Deferred processing - Schedule work to be done later

• Async operations - Inject results from other threads

• External system integration - Inject data from external sources

Example: Scheduling Deferred Work

@EventHandler
public void onNewOrder(NewOrderRequest request) {
    // Process order immediately
    processOrder(request);

    // Schedule timeout check for 30 seconds later
    OrderTimeoutCheck timeoutCheck = OrderTimeoutCheck.create();
    timeoutCheck.setOrderId(request.getOrderId());

    // Inject with 30-second delay - replicated to backup
    engine.injectMessage(timeoutCheck, 30000);
}

@EventHandler
public void onOrderTimeoutCheck(OrderTimeoutCheck check) {
    // This executes 30 seconds later on both primary and backup
    Order order = orderBook.getOrder(check.getOrderId());
    if (order != null && "PENDING".equals(order.getStatus())) {
        // Cancel timed-out order
        cancelOrder(order);
    }
}

Example: Integrating External Operations

@EventHandler
public void onHostnameRequest(BroadcastHostNameRequest request) {
    // Get hostname from local environment (non-deterministic)
    String hostname = InetAddress.getLocalHost().getHostName();

    // Inject result as a message (now deterministic)
    SendHostNameCommand command = SendHostNameCommand.create();
    command.setHostName(hostname);
    engine.injectMessage(command);
}

@EventHandler
public void onSendHostName(SendHostNameCommand command) {
    // This handler processes the injected message
    // Both primary and backup receive the same hostname
    BroadcastHostNameResponse response = BroadcastHostNameResponse.create();
    response.setHostName(command.getHostName());
    messageSender.sendMessage("hostname-response", response);
}

See Scheduling Messages for complete message injection documentation.

Environment Replication

Environment Replication allows you to tunnel local environment data into the replication stream without creating separate transactions. This is more efficient than message injection for frequently-accessed environment data.

Example: Hostname Provider

private HostNameEnvironmentProvider hostNameProvider = new HostNameEnvironmentProvider();
private XString hostName = XString.create(256, true, true);

@AppInjectionPoint
public void initialize(AepEngine engine) {
    // Register environment provider
    hostNameProvider.register(engine);
}

@EventHandler
public void onMessage(BroadcastHostNameRequest message) {
    BroadcastHostNameResponse response = BroadcastHostNameResponse.create();

    // Call environment provider - captures or replays hostname
    hostNameProvider.getHostNameTo(hostName);

    // Set hostname in response and send
    response.setHostName(hostName);
    messageSender.sendMessage("hostname-response", response);
}

How It Works:

• On primary: Provider records hostname lookup in replication buffer

• On backup: Provider replays recorded hostname from buffer

• All within same transaction - no additional message injection overhead

---

Deterministic Programming Rules

Event Sourcing requires strict adherence to deterministic programming rules:

What You MUST Do

1. Base all decisions on message data and existing state only

2. Use engine.getEngineTime() for all time-dependent logic

3. Keep handlers synchronous and single-threaded

4. Use message injection for async operations

5. Use environment replication for environment data

6. Ensure identical handler execution on all instances

What You MUST NOT Do

1. Never call System.currentTimeMillis() or System.nanoTime()

2. Never use Random or any non-deterministic algorithms

3. Never read from files, databases, or external systems directly

4. Never use thread-local storage or instance variables modified outside handlers

5. Never depend on environment variables or system properties in handlers

6. Never use object identity (e.g., System.identityHashCode()) for logic

See Programming Fundamentals for complete deterministic programming rules.

---

Related Documentation

Core Concepts

• Consensus Models - Understanding Event Sourcing vs State Replication

• Transactions - How transactions work with Event Sourcing

• Application Lifecycle - Initialization and recovery

Development

• Modeling Messages & State - ADM modeling for messages

• Programming Fundamentals - Deterministic coding rules

• Handling Messages - Writing message handlers

• Scheduling Messages - Message injection

Configuration

• Storage Configuration - Complete storage configuration reference

• Configuring Threading - Optimize replication performance

Operations

• Transaction Log Tool - Browse and analyze transaction logs

• Querying Transaction Logs - XPQL query language

---

Next Steps

1. Start modeling: Define your messages in model.xml following ADM syntax

2. Create application class: Annotate with @AppHAPolicy(EventSourcing) and design state POJOs

3. Write handlers: Implement deterministic message handlers

4. Configure storage: Enable clustering and persistence in DDL

5. Test failover: Verify state consistency through message replay

6. Test determinism: Ensure identical processing on multiple instances

7. Monitor in production: Track transaction log size and replay time