google-site-verification: google61fe8ba583a51912.html
MACRO NEPAL
Event Sourcing with Event Store in Java: Building Event-Driven Systems
awiskar acharya
November 7, 2025
4:18 pm

Article

Event Sourcing is a powerful architectural pattern that persists the state of a business entity as a sequence of state-changing events. Instead of storing just the current state, Event Sourcing stores the complete history of changes, enabling powerful capabilities like audit trails, temporal queries, and event replay. This guide explores implementing Event Sourcing with Event Store in Java applications.

Core Concepts of Event Sourcing

Key Principles:

  • Events as Source of Truth: The system state is derived from events
  • Immutable Events: Events are append-only and never modified
  • Event Replay: Current state can be rebuilt by replaying events
  • CQRS Compatibility: Often used with Command Query Responsibility Segregation

Benefits:

  • Complete Audit Trail: Every change is permanently recorded
  • Temporal Queries: Query state at any point in time
  • Event Replay: Rebuild state for new projections or bug fixes
  • Debugging: Understand exactly how state evolved

Event Store Implementation

1. Core Event Interfaces and Classes

// Base Event Interface public interface DomainEvent { String getAggregateId(); String getEventType(); Long getVersion(); Instant getTimestamp(); Map<String, Object> getMetadata(); } // Abstract Base Event public abstract class BaseEvent implements DomainEvent { private final String aggregateId; private final String eventType; private final Long version; private final Instant timestamp; private final Map<String, Object> metadata; protected BaseEvent(String aggregateId, String eventType, Long version) { this.aggregateId = aggregateId; this.eventType = eventType; this.version = version; this.timestamp = Instant.now(); this.metadata = new HashMap<>(); } // Getters @Override public String getAggregateId() { return aggregateId; } @Override public String getEventType() { return eventType; } @Override public Long getVersion() { return version; } @Override public Instant getTimestamp() { return timestamp; } @Override public Map<String, Object> getMetadata() { return metadata; } public void addMetadata(String key, Object value) { metadata.put(key, value); } } // Event Store Interface public interface EventStore { void appendEvents(String aggregateId, List<DomainEvent> events, Long expectedVersion); List<DomainEvent> getEvents(String aggregateId); List<DomainEvent> getEvents(String aggregateId, Long fromVersion); List<DomainEvent> getEventsByType(String eventType); List<DomainEvent> getEventsSince(Instant since); boolean aggregateExists(String aggregateId); }

2. Aggregate Root Base Class

// Aggregate Root Base Class public abstract class AggregateRoot { private final String id; private Long version = 0L; private final List<DomainEvent> changes = new ArrayList<>(); protected AggregateRoot(String id) { this.id = id; } public String getId() { return id; } public Long getVersion() { return version; } public List<DomainEvent> getUncommittedChanges() { return new ArrayList<>(changes); } public void markChangesAsCommitted() { changes.clear(); } protected void applyChange(DomainEvent event, boolean isNew) { try { // Use reflection to call apply method var method = getClass().getDeclaredMethod("apply", event.getClass()); method.setAccessible(true); method.invoke(this, event); if (isNew) { changes.add(event); } version++; } catch (Exception e) { throw new RuntimeException("Error applying event", e); } } protected void applyChange(DomainEvent event) { applyChange(event, true); } public void loadFromHistory(List<DomainEvent> history) { for (DomainEvent event : history) { applyChange(event, false); version = event.getVersion(); } } // Rebuild aggregate from events public static <T extends AggregateRoot> T rebuild(Class<T> aggregateClass, String aggregateId, EventStore eventStore) { try { T aggregate = aggregateClass.getDeclaredConstructor(String.class) .newInstance(aggregateId); List<DomainEvent> events = eventStore.getEvents(aggregateId); aggregate.loadFromHistory(events); return aggregate; } catch (Exception e) { throw new RuntimeException("Failed to rebuild aggregate", e); } } }

3. In-Memory Event Store Implementation

// In-Memory Event Store (for demonstration) public class InMemoryEventStore implements EventStore { private final Map<String, List<DomainEvent>> eventStore = new ConcurrentHashMap<>(); private final List<DomainEvent> allEvents = new CopyOnWriteArrayList<>(); @Override public void appendEvents(String aggregateId, List<DomainEvent> events, Long expectedVersion) { synchronized (eventStore) { List<DomainEvent> existingEvents = eventStore.getOrDefault(aggregateId, new ArrayList<>()); // Optimistic concurrency check if (expectedVersion != null && !existingEvents.isEmpty()) { Long currentVersion = existingEvents.get(existingEvents.size() - 1).getVersion(); if (!currentVersion.equals(expectedVersion)) { throw new ConcurrencyException( "Expected version " + expectedVersion + " but found " + currentVersion ); } } // Append new events List<DomainEvent> updatedEvents = new ArrayList<>(existingEvents); Long nextVersion = existingEvents.isEmpty() ? 1L : existingEvents.get(existingEvents.size() - 1).getVersion() + 1; for (DomainEvent event : events) { // Create event with correct version var versionedEvent = createVersionedEvent(event, nextVersion++); updatedEvents.add(versionedEvent); allEvents.add(versionedEvent); } eventStore.put(aggregateId, updatedEvents); } } @Override public List<DomainEvent> getEvents(String aggregateId) { return new ArrayList<>(eventStore.getOrDefault(aggregateId, new ArrayList<>())); } @Override public List<DomainEvent> getEvents(String aggregateId, Long fromVersion) { return getEvents(aggregateId).stream() .filter(event -> event.getVersion() >= fromVersion) .collect(Collectors.toList()); } @Override public List<DomainEvent> getEventsByType(String eventType) { return allEvents.stream() .filter(event -> event.getEventType().equals(eventType)) .collect(Collectors.toList()); } @Override public List<DomainEvent> getEventsSince(Instant since) { return allEvents.stream() .filter(event -> event.getTimestamp().isAfter(since)) .collect(Collectors.toList()); } @Override public boolean aggregateExists(String aggregateId) { return eventStore.containsKey(aggregateId); } private DomainEvent createVersionedEvent(DomainEvent event, Long version) { // In a real implementation, you'd create a new event instance with the version // For simplicity, we assume events are mutable in this example try { var constructor = event.getClass().getDeclaredConstructor( String.class, String.class, Long.class); constructor.setAccessible(true); return constructor.newInstance( event.getAggregateId(), event.getEventType(), version ); } catch (Exception e) { throw new RuntimeException("Failed to create versioned event", e); } } } // Custom exception for concurrency conflicts public class ConcurrencyException extends RuntimeException { public ConcurrencyException(String message) { super(message); } }

Bank Account Example

1. Bank Account Aggregate

// Bank Account Aggregate public class BankAccount extends AggregateRoot { private String accountNumber; private String accountHolder; private BigDecimal balance; private AccountStatus status; public BankAccount(String id) { super(id); } // Command methods public static BankAccount create(String accountId, String accountHolder, String accountNumber) { BankAccount account = new BankAccount(accountId); account.applyChange(new AccountCreatedEvent(accountId, accountHolder, accountNumber)); return account; } public void deposit(BigDecimal amount) { if (status != AccountStatus.ACTIVE) { throw new IllegalStateException("Account is not active"); } if (amount.compareTo(BigDecimal.ZERO) <= 0) { throw new IllegalArgumentException("Deposit amount must be positive"); } applyChange(new MoneyDepositedEvent(getId(), amount, balance.add(amount))); } public void withdraw(BigDecimal amount) { if (status != AccountStatus.ACTIVE) { throw new IllegalStateException("Account is not active"); } if (amount.compareTo(BigDecimal.ZERO) <= 0) { throw new IllegalArgumentException("Withdrawal amount must be positive"); } if (balance.compareTo(amount) < 0) { throw new IllegalArgumentException("Insufficient funds"); } applyChange(new MoneyWithdrawnEvent(getId(), amount, balance.subtract(amount))); } public void closeAccount() { if (status != AccountStatus.ACTIVE) { throw new IllegalStateException("Account is not active"); } applyChange(new AccountClosedEvent(getId())); } // Event application methods @SuppressWarnings("unused") private void apply(AccountCreatedEvent event) { this.accountNumber = event.getAccountNumber(); this.accountHolder = event.getAccountHolder(); this.balance = BigDecimal.ZERO; this.status = AccountStatus.ACTIVE; } @SuppressWarnings("unused") private void apply(MoneyDepositedEvent event) { this.balance = event.getNewBalance(); } @SuppressWarnings("unused") private void apply(MoneyWithdrawnEvent event) { this.balance = event.getNewBalance(); } @SuppressWarnings("unused") private void apply(AccountClosedEvent event) { this.status = AccountStatus.CLOSED; } // Getters public String getAccountNumber() { return accountNumber; } public String getAccountHolder() { return accountHolder; } public BigDecimal getBalance() { return balance; } public AccountStatus getStatus() { return status; } } // Account Status Enum enum AccountStatus { ACTIVE, CLOSED, SUSPENDED }

2. Bank Account Events

// Account Events public class AccountCreatedEvent extends BaseEvent { private final String accountHolder; private final String accountNumber; public AccountCreatedEvent(String aggregateId, String accountHolder, String accountNumber) { super(aggregateId, "ACCOUNT_CREATED", 1L); this.accountHolder = accountHolder; this.accountNumber = accountNumber; } public String getAccountHolder() { return accountHolder; } public String getAccountNumber() { return accountNumber; } } public class MoneyDepositedEvent extends BaseEvent { private final BigDecimal amount; private final BigDecimal newBalance; public MoneyDepositedEvent(String aggregateId, BigDecimal amount, BigDecimal newBalance) { super(aggregateId, "MONEY_DEPOSITED", null); this.amount = amount; this.newBalance = newBalance; } public BigDecimal getAmount() { return amount; } public BigDecimal getNewBalance() { return newBalance; } } public class MoneyWithdrawnEvent extends BaseEvent { private final BigDecimal amount; private final BigDecimal newBalance; public MoneyWithdrawnEvent(String aggregateId, BigDecimal amount, BigDecimal newBalance) { super(aggregateId, "MONEY_WITHDRAWN", null); this.amount = amount; this.newBalance = newBalance; } public BigDecimal getAmount() { return amount; } public BigDecimal getNewBalance() { return newBalance; } } public class AccountClosedEvent extends BaseEvent { public AccountClosedEvent(String aggregateId) { super(aggregateId, "ACCOUNT_CLOSED", null); } }

Repository and Command Handling

1. Repository Implementation

// Generic Repository public class EventSourcingRepository<T extends AggregateRoot> { private final EventStore eventStore; private final Class<T> aggregateClass; public EventSourcingRepository(EventStore eventStore, Class<T> aggregateClass) { this.eventStore = eventStore; this.aggregateClass = aggregateClass; } public void save(T aggregate) { List<DomainEvent> changes = aggregate.getUncommittedChanges(); Long expectedVersion = aggregate.getVersion() - changes.size(); eventStore.appendEvents(aggregate.getId(), changes, expectedVersion); aggregate.markChangesAsCommitted(); } public T findById(String id) { if (!eventStore.aggregateExists(id)) { return null; } return AggregateRoot.rebuild(aggregateClass, id, eventStore); } public T getById(String id) { T aggregate = findById(id); if (aggregate == null) { throw new AggregateNotFoundException("Aggregate not found: " + id); } return aggregate; } } public class AggregateNotFoundException extends RuntimeException { public AggregateNotFoundException(String message) { super(message); } } // Bank Account Repository public class BankAccountRepository extends EventSourcingRepository<BankAccount> { public BankAccountRepository(EventStore eventStore) { super(eventStore, BankAccount.class); } }

2. Command Handling

// Commands public interface Command { String getAggregateId(); } public class CreateAccountCommand implements Command { private final String accountId; private final String accountHolder; private final String accountNumber; public CreateAccountCommand(String accountId, String accountHolder, String accountNumber) { this.accountId = accountId; this.accountHolder = accountHolder; this.accountNumber = accountNumber; } @Override public String getAggregateId() { return accountId; } public String getAccountHolder() { return accountHolder; } public String getAccountNumber() { return accountNumber; } } public class DepositMoneyCommand implements Command { private final String accountId; private final BigDecimal amount; public DepositMoneyCommand(String accountId, BigDecimal amount) { this.accountId = accountId; this.amount = amount; } @Override public String getAggregateId() { return accountId; } public BigDecimal getAmount() { return amount; } } public class WithdrawMoneyCommand implements Command { private final String accountId; private final BigDecimal amount; public WithdrawMoneyCommand(String accountId, BigDecimal amount) { this.accountId = accountId; this.amount = amount; } @Override public String getAggregateId() { return accountId; } public BigDecimal getAmount() { return amount; } } // Command Handler public class BankAccountCommandHandler { private final BankAccountRepository repository; public BankAccountCommandHandler(BankAccountRepository repository) { this.repository = repository; } public void handle(CreateAccountCommand command) { BankAccount account = BankAccount.create( command.getAggregateId(), command.getAccountHolder(), command.getAccountNumber() ); repository.save(account); } public void handle(DepositMoneyCommand command) { BankAccount account = repository.getById(command.getAggregateId()); account.deposit(command.getAmount()); repository.save(account); } public void handle(WithdrawMoneyCommand command) { BankAccount account = repository.getById(command.getAggregateId()); account.withdraw(command.getAmount()); repository.save(account); } }

Projections and Read Models

1. Projection System

// Projection Interface public interface Projection { void handle(DomainEvent event); String getProjectionName(); } // Event Processor public class EventProcessor { private final EventStore eventStore; private final List<Projection> projections; public EventProcessor(EventStore eventStore) { this.eventStore = eventStore; this.projections = new ArrayList<>(); } public void registerProjection(Projection projection) { projections.add(projection); } public void processEvent(DomainEvent event) { for (Projection projection : projections) { try { projection.handle(event); } catch (Exception e) { // Log error but continue processing other projections System.err.println("Error processing event in projection " + projection.getProjectionName() + ": " + e.getMessage()); } } } public void replayAllEvents() { List<DomainEvent> allEvents = eventStore.getEventsSince(Instant.EPOCH); for (DomainEvent event : allEvents) { processEvent(event); } } public void replayEventsSince(Instant since) { List<DomainEvent> events = eventStore.getEventsSince(since); for (DomainEvent event : events) { processEvent(event); } } }

2. Bank Account Projections

// Account Summary Projection public class AccountSummaryProjection implements Projection { private final Map<String, AccountSummary> accountSummaries = new ConcurrentHashMap<>(); @Override public void handle(DomainEvent event) { switch (event.getEventType()) { case "ACCOUNT_CREATED": handle((AccountCreatedEvent) event); break; case "MONEY_DEPOSITED": handle((MoneyDepositedEvent) event); break; case "MONEY_WITHDRAWN": handle((MoneyWithdrawnEvent) event); break; case "ACCOUNT_CLOSED": handle((AccountClosedEvent) event); break; } } private void handle(AccountCreatedEvent event) { accountSummaries.put(event.getAggregateId(), new AccountSummary(event.getAggregateId(), event.getAccountHolder(), event.getAccountNumber(), BigDecimal.ZERO, AccountStatus.ACTIVE)); } private void handle(MoneyDepositedEvent event) { AccountSummary summary = accountSummaries.get(event.getAggregateId()); if (summary != null) { accountSummaries.put(event.getAggregateId(), new AccountSummary(summary.getAccountId(), summary.getAccountHolder(), summary.getAccountNumber(), event.getNewBalance(), summary.getStatus())); } } private void handle(MoneyWithdrawnEvent event) { AccountSummary summary = accountSummaries.get(event.getAggregateId()); if (summary != null) { accountSummaries.put(event.getAggregateId(), new AccountSummary(summary.getAccountId(), summary.getAccountHolder(), summary.getAccountNumber(), event.getNewBalance(), summary.getStatus())); } } private void handle(AccountClosedEvent event) { AccountSummary summary = accountSummaries.get(event.getAggregateId()); if (summary != null) { accountSummaries.put(event.getAggregateId(), new AccountSummary(summary.getAccountId(), summary.getAccountHolder(), summary.getAccountNumber(), summary.getBalance(), AccountStatus.CLOSED)); } } @Override public String getProjectionName() { return "AccountSummaryProjection"; } public AccountSummary getAccountSummary(String accountId) { return accountSummaries.get(accountId); } public List<AccountSummary> getAllAccountSummaries() { return new ArrayList<>(accountSummaries.values()); } } // Account Summary DTO public class AccountSummary { private final String accountId; private final String accountHolder; private final String accountNumber; private final BigDecimal balance; private final AccountStatus status; public AccountSummary(String accountId, String accountHolder, String accountNumber, BigDecimal balance, AccountStatus status) { this.accountId = accountId; this.accountHolder = accountHolder; this.accountNumber = accountNumber; this.balance = balance; this.status = status; } // Getters public String getAccountId() { return accountId; } public String getAccountHolder() { return accountHolder; } public String getAccountNumber() { return accountNumber; } public BigDecimal getBalance() { return balance; } public AccountStatus getStatus() { return status; } }

Database-Backed Event Store

1. PostgreSQL Event Store Implementation

// PostgreSQL Event Store public class PostgresEventStore implements EventStore { private final DataSource dataSource; private final ObjectMapper objectMapper; public PostgresEventStore(DataSource dataSource) { this.dataSource = dataSource; this.objectMapper = new ObjectMapper(); initializeSchema(); } private void initializeSchema() { String sql = """ CREATE TABLE IF NOT EXISTS events ( id BIGSERIAL PRIMARY KEY, aggregate_id VARCHAR(255) NOT NULL, event_type VARCHAR(255) NOT NULL, event_data JSONB NOT NULL, version BIGINT NOT NULL, timestamp TIMESTAMP NOT NULL, metadata JSONB, UNIQUE(aggregate_id, version) ); CREATE INDEX IF NOT EXISTS idx_events_aggregate_id ON events(aggregate_id); CREATE INDEX IF NOT EXISTS idx_events_event_type ON events(event_type); CREATE INDEX IF NOT EXISTS idx_events_timestamp ON events(timestamp); """; try (Connection conn = dataSource.getConnection(); Statement stmt = conn.createStatement()) { stmt.execute(sql); } catch (SQLException e) { throw new RuntimeException("Failed to initialize event store schema", e); } } @Override public void appendEvents(String aggregateId, List<DomainEvent> events, Long expectedVersion) { String sql = """ INSERT INTO events (aggregate_id, event_type, event_data, version, timestamp, metadata) VALUES (?, ?, ?::jsonb, ?, ?, ?::jsonb) """; try (Connection conn = dataSource.getConnection(); PreparedStatement stmt = conn.prepareStatement(sql)) { conn.setAutoCommit(false); // Check current version if expected version is provided if (expectedVersion != null) { Long currentVersion = getCurrentVersion(conn, aggregateId); if (!currentVersion.equals(expectedVersion)) { throw new ConcurrencyException( "Expected version " + expectedVersion + " but found " + currentVersion ); } } Long nextVersion = getNextVersion(conn, aggregateId); for (DomainEvent event : events) { stmt.setString(1, aggregateId); stmt.setString(2, event.getEventType()); stmt.setString(3, objectMapper.writeValueAsString(event)); stmt.setLong(4, nextVersion++); stmt.setTimestamp(5, Timestamp.from(event.getTimestamp())); stmt.setString(6, objectMapper.writeValueAsString(event.getMetadata())); stmt.addBatch(); } stmt.executeBatch(); conn.commit(); } catch (Exception e) { throw new RuntimeException("Failed to append events", e); } } @Override public List<DomainEvent> getEvents(String aggregateId) { String sql = "SELECT event_data FROM events WHERE aggregate_id = ? ORDER BY version"; try (Connection conn = dataSource.getConnection(); PreparedStatement stmt = conn.prepareStatement(sql)) { stmt.setString(1, aggregateId); ResultSet rs = stmt.executeQuery(); List<DomainEvent> events = new ArrayList<>(); while (rs.next()) { String eventData = rs.getString("event_data"); // Deserialize based on event_type DomainEvent event = deserializeEvent(eventData); events.add(event); } return events; } catch (Exception e) { throw new RuntimeException("Failed to get events", e); } } private Long getCurrentVersion(Connection conn, String aggregateId) throws SQLException { String sql = "SELECT COALESCE(MAX(version), 0) as current_version FROM events WHERE aggregate_id = ?"; try (PreparedStatement stmt = conn.prepareStatement(sql)) { stmt.setString(1, aggregateId); ResultSet rs = stmt.executeQuery(); return rs.next() ? rs.getLong("current_version") : 0L; } } private Long getNextVersion(Connection conn, String aggregateId) throws SQLException { return getCurrentVersion(conn, aggregateId) + 1; } private DomainEvent deserializeEvent(String eventData) throws Exception { JsonNode jsonNode = objectMapper.readTree(eventData); String eventType = jsonNode.get("eventType").asText(); // Map event type to class Class<? extends DomainEvent> eventClass = switch (eventType) { case "ACCOUNT_CREATED" -> AccountCreatedEvent.class; case "MONEY_DEPOSITED" -> MoneyDepositedEvent.class; case "MONEY_WITHDRAWN" -> MoneyWithdrawnEvent.class; case "ACCOUNT_CLOSED" -> AccountClosedEvent.class; default -> throw new IllegalArgumentException("Unknown event type: " + eventType); }; return objectMapper.readValue(eventData, eventClass); } // Other methods implementation similar to InMemoryEventStore @Override public List<DomainEvent> getEvents(String aggregateId, Long fromVersion) { // Implementation similar to getEvents with version filter return List.of(); } @Override public List<DomainEvent> getEventsByType(String eventType) { // Implementation return List.of(); } @Override public List<DomainEvent> getEventsSince(Instant since) { // Implementation return List.of(); } @Override public boolean aggregateExists(String aggregateId) { // Implementation return false; } }

Complete Usage Example

// Complete Example public class EventSourcingExample { public static void main(String[] args) { // Setup EventStore eventStore = new InMemoryEventStore(); BankAccountRepository repository = new BankAccountRepository(eventStore); BankAccountCommandHandler commandHandler = new BankAccountCommandHandler(repository); // Setup projections EventProcessor eventProcessor = new EventProcessor(eventStore); AccountSummaryProjection accountSummaryProjection = new AccountSummaryProjection(); eventProcessor.registerProjection(accountSummaryProjection); // Process some commands String accountId = "acc-123"; // Create account commandHandler.handle(new CreateAccountCommand(accountId, "John Doe", "ACC001")); // Deposit money commandHandler.handle(new DepositMoneyCommand(accountId, new BigDecimal("1000.00"))); // Withdraw money commandHandler.handle(new WithdrawMoneyCommand(accountId, new BigDecimal("200.00"))); // Replay events to build projections eventProcessor.replayAllEvents(); // Query projections AccountSummary summary = accountSummaryProjection.getAccountSummary(accountId); System.out.println("Account Balance: " + summary.getBalance()); // Rebuild aggregate from events BankAccount rebuiltAccount = repository.findById(accountId); System.out.println("Rebuilt Balance: " + rebuiltAccount.getBalance()); // Show event history List<DomainEvent> history = eventStore.getEvents(accountId); System.out.println("Event History:"); history.forEach(event -> System.out.println(" - " + event.getEventType() + " v" + event.getVersion()) ); } }

Best Practices

  1. Event Versioning: Plan for event schema evolution
  2. Idempotent Handlers: Ensure event processing is idempotent
  3. Snapshot Strategy: Implement snapshots for large aggregates
  4. Event Serialization: Use backward-compatible serialization formats
  5. Error Handling: Implement robust error handling in projections
  6. Monitoring: Track event counts, processing latency, and errors
  7. Testing: Thoroughly test event replay and projection rebuilding

Conclusion

Event Sourcing with Event Store provides a robust foundation for building event-driven systems in Java. By storing all state changes as immutable events, you gain complete auditability, temporal query capabilities, and the ability to rebuild state from scratch. The pattern works exceptionally well with CQRS and enables powerful features like event replay and multiple read models. While it introduces complexity in event schema management and requires careful consideration of performance for large event streams, the benefits for complex business domains often outweigh the costs. Proper implementation with the patterns shown here will result in a maintainable, scalable, and flexible system architecture.

Category : JAVA

Leave a Reply

Your email address will not be published. Required fields are marked *


Macro Nepal Helper