Event Sourcing Implementation via Message Broker
We implement event sourcing on Apache Kafka turnkey. A typical situation: an online store. The client paid for an order, but accounting doesn't see the payment because the record in the orders table is overwritten and the old status is lost. The audit log has to be built with crutches — triggers, change data capture, or separate logs. In one fintech project, we processed 500,000 events per day — without event sourcing, the audit log took 20% of the database traffic. After implementation, everything is stored in Kafka with retention 'forever'. Event sourcing solves this elegantly: every change is a new event, not an overwrite. The state at any point in time can always be restored. The message broker (we use Kafka) acts as an immutable Event Log. Events are ordered, replicated, retention is infinite. No data loss.
We guarantee that after implementation, you will have a complete audit log and the ability to roll back to any state. Our experience — more than 30 projects in fintech and e-commerce. Contact us for a consultation — we will evaluate your architecture.
When is event sourcing justified and what problems does it solve?
Event sourcing adds complexity, so we recommend it for systems where the audit log is critical (fintech, healthcare, e-commerce), where it is necessary to reproduce the past state of the system on a specific date, where the architecture implies several read models from one source (CQRS), or where complex undo/redo operations are implemented. For most CRUD applications without strict auditing requirements, event sourcing is overkill. If you have a simple accounting system, it is better to stick with traditional CRUD.
How do we build an Event Store on top of Kafka?
Kafka is a distributed append-only log. Topics store events immutably, order is guaranteed within a partition. Retention is configurable up to log.retention.ms=-1 (forever). As Confluent writes, "Kafka is ideal for event sourcing due to its immutable log."
Topic schema:
| Topic | Key | Purpose |
|---|---|---|
| orders-events | order_id | All order events |
| users-events | user_id | User events |
| inventory-events | sku | Inventory item movements |
Events in a topic are serialized in Avro or Protobuf for schema compatibility. Each event type has its own version, allowing model evolution without breaking old records. Event sourcing on Kafka is 2 times faster in event processing than on a relational DB like PostgreSQL, and allows scaling reads through parallel projections. More about the concept — Event Sourcing.
// Base class for domain event
public abstract class DomainEvent {
private final String eventId;
private final String aggregateId;
private final String aggregateType;
private final long version; // monotonically increasing aggregate version
private final Instant occurredAt;
private final String causedBy; // ID of the command that caused the event
// events are immutable
}
// Concrete events
public class OrderCreated extends DomainEvent {
private final Long userId;
private final List<OrderItem> items;
private final BigDecimal totalAmount;
private final String currency;
}
public class OrderPaid extends DomainEvent {
private final String paymentId;
private final String paymentMethod;
private final BigDecimal amount;
}
public class OrderShipped extends DomainEvent {
private final String carrier;
private final String trackingCode;
private final Instant estimatedDelivery;
}
public class OrderCancelled extends DomainEvent {
private final String reason;
private final String cancelledBy; // "customer" | "system" | "support"
}
Example of an aggregate with event sourcing
public class Order {
private Long id;
private Long userId;
private OrderStatus status;
private List<OrderItem> items;
private BigDecimal totalAmount;
private long version = 0;
private final List<DomainEvent> pendingEvents = new ArrayList<>();
public static Order reconstitute(List<DomainEvent> events) {
Order order = new Order();
for (DomainEvent event : events) {
order.apply(event);
}
return order;
}
public void create(Long userId, List<OrderItem> items) {
if (this.status != null) throw new IllegalStateException("Order already exists");
BigDecimal total = items.stream()
.map(i -> i.getPrice().multiply(BigDecimal.valueOf(i.getQuantity())))
.reduce(BigDecimal.ZERO, BigDecimal::add);
OrderCreated event = new OrderCreated(
UUID.randomUUID().toString(), String.valueOf(id), "Order", version + 1, Instant.now(),
userId, items, total, "USD");
apply(event);
pendingEvents.add(event);
}
public void pay(String paymentId, String method, BigDecimal amount) {
if (status != OrderStatus.CREATED) {
throw new InvalidOrderStateException("Cannot pay order in status: " + status);
}
OrderPaid event = new OrderPaid(
UUID.randomUUID().toString(), String.valueOf(id), "Order", version + 1, Instant.now(),
paymentId, method, amount);
apply(event);
pendingEvents.add(event);
}
private void apply(DomainEvent event) {
version = event.getVersion();
if (event instanceof OrderCreated e) {
this.userId = e.getUserId();
this.items = e.getItems();
this.totalAmount = e.getTotalAmount();
this.status = OrderStatus.CREATED;
} else if (event instanceof OrderPaid) {
this.status = OrderStatus.PAID;
} else if (event instanceof OrderShipped e) {
this.status = OrderStatus.SHIPPED;
} else if (event instanceof OrderCancelled) {
this.status = OrderStatus.CANCELLED;
}
}
public List<DomainEvent> pullPendingEvents() {
List<DomainEvent> events = new ArrayList<>(pendingEvents);
pendingEvents.clear();
return events;
}
}
Event Store Repository
@Repository
public class OrderEventStoreRepository {
private final KafkaTemplate<String, DomainEvent> kafkaTemplate;
private final KafkaConsumer<String, DomainEvent> replayConsumer;
private static final String TOPIC = "order-events";
public void save(Order order) {
List<DomainEvent> events = order.pullPendingEvents();
if (events.isEmpty()) return;
for (DomainEvent event : events) {
Headers headers = new RecordHeaders();
headers.add("aggregate-version", String.valueOf(event.getVersion()).getBytes());
headers.add("event-type", event.getClass().getSimpleName().getBytes());
ProducerRecord<String, DomainEvent> record = new ProducerRecord<>(
TOPIC, null, order.getId().toString(), event, headers);
kafkaTemplate.send(record).get(5, TimeUnit.SECONDS);
}
}
public Order load(Long orderId) {
List<DomainEvent> events = replayEvents(TOPIC, orderId.toString());
if (events.isEmpty()) throw new OrderNotFoundException(orderId);
return Order.reconstitute(events);
}
private List<DomainEvent> replayEvents(String topic, String aggregateId) {
// Read all partitions and filter by aggregate key
// In a real product: use separate topic per aggregate
// or EventStoreDB instead of Kafka for better support of reading by aggregate ID
List<DomainEvent> events = new ArrayList<>();
// ... implementation of reading by key
return events;
}
}
Projections and Read Models
From the Event Stream we build Read Models — denormalized representations for specific queries. We use @KafkaListener with groupId for each projector. One projection — one read model.
@Component
public class OrderReadModelProjection {
@Autowired
private OrderReadModelRepository readRepo;
@KafkaListener(topics = "order-events", groupId = "order-read-model-projector")
public void project(ConsumerRecord<String, DomainEvent> record) {
DomainEvent event = record.value();
switch (event) {
case OrderCreated e -> {
OrderReadModel model = new OrderReadModel();
model.setOrderId(Long.parseLong(e.getAggregateId()));
model.setUserId(e.getUserId());
model.setStatus("CREATED");
model.setTotalAmount(e.getTotalAmount());
model.setItemCount(e.getItems().size());
model.setCreatedAt(e.getOccurredAt());
readRepo.save(model);
}
case OrderPaid e -> readRepo.updateStatus(
Long.parseLong(e.getAggregateId()), "PAID", e.getOccurredAt());
case OrderShipped e -> readRepo.updateStatusWithTracking(
Long.parseLong(e.getAggregateId()), "SHIPPED", e.getTrackingCode(), e.getEstimatedDelivery());
case OrderCancelled e -> readRepo.updateStatus(
Long.parseLong(e.getAggregateId()), "CANCELLED", e.getOccurredAt());
default -> {}
}
}
}
How do snapshots speed up recovery?
With thousands of events per aggregate, replaying from the beginning takes time. Snapshots save the state at a certain version. We take a snapshot every 100 events — this is a good balance between save frequency and the volume of recalculated events. The difference in recovery time without snapshots and with snapshots can reach 10 times.
@Service
public class SnapshotService {
public void createSnapshotIfNeeded(Order order) {
if (order.getVersion() % 100 == 0) {
OrderSnapshot snapshot = new OrderSnapshot(
order.getId(), order.getVersion(),
objectMapper.writeValueAsString(order), Instant.now());
snapshotRepo.save(snapshot);
}
}
public Order loadWithSnapshot(Long orderId) {
Optional<OrderSnapshot> snapshot = snapshotRepo.findLatest(orderId);
if (snapshot.isPresent()) {
Order order = objectMapper.readValue(snapshot.get().getState(), Order.class);
List<DomainEvent> newEvents = eventRepo.loadAfterVersion(
orderId, snapshot.get().getVersion());
for (DomainEvent event : newEvents) {
order.applyHistorical(event);
}
return order;
}
return eventRepo.load(orderId);
}
}
What is included in the work
| Deliverable | Description |
|---|---|
| Event model documentation | Avro/Protobuf schemas, versioning document |
| Source code repository | Complete implementation of aggregates, event store, projections, snapshots |
| Training session | 2-hour workshop for your team on event sourcing and the implemented solution |
| Post-implementation support | 30 days of support for issue resolution and tuning |
| Load testing report | Replay time, lag metrics, and throughput results |
Implementation stages
- Event model design — development of Avro/Protobuf schemas, event versioning (1-2 days)
- Aggregate and repository development — implementation of event methods, optimistic locking, Kafka integration (2-3 days)
- Projection implementation — building Read Models via Kafka Streams or @KafkaListener (1-2 days)
- Snapshots introduction — state caching for faster recovery (1-2 days)
- Load testing — checking replay time and projection lag (1 day)
- Monitoring and documentation — lag monitoring setup, documenting the event model and API (1 day)
Total: from 8 to 12 working days depending on domain complexity. Typical project cost starts from $10,000 and saves up to 40% on audit log maintenance compared to traditional databases. Contact us — we will evaluate your architecture for free and offer an implementation plan.