§ learn

Consuming events

After your first Action commits, eventlog.events contains a durable event row. From there you choose how to use it: handle it locally, publish from a handler, stream with CDC, or fan out through a router.

Ekbatan’s main job is to persist events safely. When an Action commits, the domain rows and the eventlog.events rows are written in the same database transaction. After that, your application chooses how to use those persisted events.

This page shows the main ways to consume those event rows: handle them inside the same service, publish them from a handler, stream them with CDC, or route them into multiple broker topics.

                         eventlog.events
                               │
             ┌─────────────────┴──────────────────┐
             ▼                                    ▼
   local-event-handler                      Debezium CDC
   fan-out + handling jobs                  WAL/binlog -> raw topic
             │                                    │
      handler code                             router / stream app
             │                                    │
   local work or broker publish          derived topics / rekeyed streams

You do not have to pick only one. Local handlers and CDC can run simultaneously against the same events table. The local handler’s delivered=true flip generates an UPDATE row that the Debezium SMTs drop, so only the original INSERT rows ship to Kafka.

Choose a delivery topology

1. Local only — listen to yourself

Use local-event-handler when the consumer is part of the same application deployment.

eventlog.events -> EventFanoutJob -> event_notifications -> EventHandlingJob -> handler

This is the simplest path. There is no broker, no CDC connector, and no separate consumer application. It is a good fit for local projections, notification rows, audit rows, internal workflows, saga steps inside the same service, and “after this action commits, do this other piece of application work”.

You write an event handler for one event type that is being saved in the database. Whenever a new event of that type is persisted, the local event handler job picks it up and invokes your handler.

The handler receives an EventEnvelope<E> with the typed event payload, action metadata, and source action params. It can write a projection, call another service, or dispatch another Ekbatan action through ActionExecutor.

For the full API, retry behavior, and delivery table details, see the local-event-listener reference.

2. Local handler as a broker publisher

You can write an event handler for an event, and then in that handler manually interact with Kafka, Pulsar, RabbitMQ, SQS, or another broker.

eventlog.events -> EventHandlingJob -> your handler -> Kafka / Pulsar / RabbitMQ / SQS / another broker

Think of the handler as a small publisher worker. You create a local event handler, listen to the local event, and then manually publish it yourself to Kafka, Pulsar, RabbitMQ, SQS, or another message broker. Ekbatan first commits the database change and the event row; later, the handling job invokes your handler, and your handler decides what broker messages to send.

Choose this shape when publishing is not a simple “copy each event to one topic” operation. Because you publish in application code, you decide the topic or stream, message key, headers, schema, partitioning key, tenant route, and destination broker. A single WalletMoneyDepositedEvent could publish one message to wallet.deposits, another to risk.large-deposits, and nothing to billing if the action params say the deposit was internal.

It is also a good fit when the outgoing message needs application-specific enrichment before it leaves the service: for example, adding tenant metadata, mapping an internal event into a public integration event, redacting fields, or publishing different contracts for different consumers.

The tradeoff is that the handler is now an at-least-once publisher. If the broker publish succeeds but the handler process crashes before the notification row is marked SUCCEEDED, the handler can run again. Use the event row id, or another stable business key, so downstream consumers can dedupe.

3. CDC directly to one raw broker topic

Use Debezium or another CDC tool when you want infrastructure to stream committed event rows without application code publishing to the broker.

eventlog.events -> Debezium -> ekbatan.<namespace>

This produces a raw event stream for the service. The row in eventlog.events is the source of truth; the broker topic is a projection of that table. Consumers can subscribe to the raw topic and filter by model_type, event_type, namespace, tenant, or any metadata you include in the message envelope.

For Kafka, use model_id as the message key when consumers need per-aggregate ordering. That keeps events for the same wallet, order, account, or other model instance on the same partition.

4. CDC to raw topic, then fan out or rekey

Use this when “one queue” is not enough.

eventlog.events -> Debezium -> raw topic -> router / stream processor -> derived topics

The CDC connector writes every event to a raw topic. A downstream routing layer then branches the stream into whatever shape the organization needs:

one topic per event_type
one topic per model_type
one topic per tenant, region, or bounded context
multiple topics for the same event
a rekeyed topic where the partition key is changed for a specific consumer

Good tools for this layer include Kafka Streams, ksqlDB, Apache Flink, Apache Beam, Pulsar Functions, or a small custom consumer/producer service. The important distinction is that fan-out and repartitioning happen after the durable event stream exists; Ekbatan does not force the broker topology into the database write path.

Path A — Local event handler (start here)

The fastest path: no broker, no CDC, no Kafka cluster. The ekbatan-events:local-event-handler module ships two DistributedJobs — EventFanoutJob materializes per-handler delivery rows in event_notifications; EventHandlingJob polls those rows and calls your typed handler beans. Both jobs are cluster-exclusive (one cluster member runs each); the handler runs in whichever JVM is currently running EventHandlingJob, with retries, backoff, and dead-lettering handled automatically.

A.1 Add the module

If you use the Spring Boot starter, Quarkus extension, or Micronaut module, the local event handler and distributed jobs runtime are already on the classpath.

If you wire Ekbatan manually, add the local event handler module:

Gradle:

dependencies {
    implementation("io.github.zyraz-io:ekbatan-local-event-handler:0.2.1")
}

Maven:

<dependency>
  <groupId>io.github.zyraz-io</groupId>
  <artifactId>ekbatan-local-event-handler</artifactId>
  <version>0.2.1</version>
</dependency>

The local event handler module depends on ekbatan-distributed-jobs, because the fan-out and handling loops run as distributed jobs.

A.2 Write the handler

package io.example.wallet.handler;

import io.ekbatan.di.EkbatanEventHandler;
import io.ekbatan.events.localeventhandler.EventEnvelope;
import io.ekbatan.events.localeventhandler.EventHandler;
import io.example.wallet.model.events.WalletMoneyDepositedEvent;

@EkbatanEventHandler
public class WalletMoneyDepositedNotifier implements EventHandler<WalletMoneyDepositedEvent> {

    @Override
    public String name() {
        // Cluster-stable identifier; stored in event_notifications.handler_name.
        // If you rename it, keep the old value in aliases() until old rows drain.
        return "wallet-money-deposited-notifier";
    }

    @Override
    public Class<WalletMoneyDepositedEvent> eventType() {
        return WalletMoneyDepositedEvent.class;
    }

    @Override
    public void handle(EventEnvelope<WalletMoneyDepositedEvent> envelope) {
        final var event = envelope.event;
        System.out.printf("Deposited %s to wallet %s — new balance %s%n",
                event.amount, event.modelId, event.newBalance);
        // For real apps: dispatch another action via injected ActionExecutor,
        // call an SMS API, write a projection, etc.
    }
}

Three required methods, plus one optional rename helper:

name() — cluster-stable handler identifier. Stored in event_notifications.handler_name. Don’t rename casually — changing it makes future notifications use a new handler identity, while queued rows under the old name only resolve if that old name is listed in aliases(). Already fanned-out historical events are not replayed for the new name; only source events still marked delivered=false can be backfilled.
aliases() — optional previous names for the same logical handler. Use this when you rename a handler but still need to process rows that were already written with the old handler_name. The handling job can resolve old rows through aliases(), but the fan-out job writes only the current name() for new rows.
eventType() — what payload class this handler accepts. Drives JSON deserialization.
handle(envelope) — your business logic. envelope.event is the typed payload; envelope.actionParams is the JSON of the source action’s params (handy for routing concerns you don’t want polluting the event payload).

For example, suppose the handler used to return "wallet-money-deposited-notifier", and you rename the durable subscription to "wallet-money-deposit-alert". Existing event_notifications rows may still contain the old name, so keep it as an alias:

@Override
public String name() {
    return "wallet-money-deposit-alert";
}

@Override
public Set<String> aliases() {
    return Set.of("wallet-money-deposited-notifier");
}

After that deploy, old rows with handler_name = 'wallet-money-deposited-notifier' still invoke this handler through aliases(), and new rows are created with handler_name = 'wallet-money-deposit-alert'. Remove the alias only after the old-name queue has drained, or keep it indefinitely if old rows might reappear from restored backups or delayed environments.

The Local event handler reference has an animated walkthrough that shows this rename mechanism: old rows routing through aliases(), and new rows routing through the current name().

Per the listen-to-yourself pattern, the handler can dispatch its own actions via an injected ActionExecutor, and those actions persist their own resulting events through the normal action flow. That same shape is used for sagas when one committed action should trigger the next committed action.

A.3 Run

Restart your app. The next deposit you POST will produce both a WalletMoneyDepositedEvent row and a console line from the handler within ~1s (the polling interval). Check eventlog.event_notifications to see the handler-specific delivery record:

SELECT event_id, handler_name, state, attempts, updated_date
FROM eventlog.event_notifications
WHERE handler_name = 'wallet-money-deposited-notifier'
ORDER BY created_date DESC;

state = 'SUCCEEDED' means the handler returned without throwing. Retries, backoff, expiry, and dead-lettering all happen automatically — see Local event handler.

Path B — Debezium CDC → Kafka

When you need fan-out beyond your own application (other services, a data lake, an analytics warehouse), tail eventlog.events with Debezium. The framework ships two Kafka Connect SMTs that can turn real event rows into Avro or Protobuf bytes before they reach Kafka. A downstream router can then fan the raw topic out to per-model or per-event topics.

B.1 Set up Debezium

Point a Debezium Postgres source connector at the database. Capture only the events table:

{
  "name": "wallet-events",
  "config": {
    "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
    "tasks.max": "1",
    "database.hostname": "postgres",
    "database.port": "5432",
    "database.user": "wallet",
    "database.password": "wallet",
    "database.dbname": "wallet",
    "schema.include.list": "eventlog",
    "table.include.list": "eventlog.events",
    "plugin.name": "pgoutput",

    "value.converter": "org.apache.kafka.connect.converters.ByteArrayConverter",

    "transforms": "encodeAvro",
    "transforms.encodeAvro.type": "io.ekbatan.events.streaming.debeziumsmt.avro.OutboxToAvroTransform",
    "transforms.encodeAvro.actionEventSchema": "/schemas/ActionEvent.avsc",
    "transforms.encodeAvro.payloadSchemas": "WalletMoneyDepositedEvent:/schemas/WalletMoneyDepositedEvent.avsc",

    "errors.retry.timeout": "600000",
    "errors.retry.delay.max.ms": "30000",
    "errors.tolerance": "all",
    "errors.log.enable": "true",
    "errors.log.include.messages": "false"
  }
}

The OutboxToAvroTransform:

Drops UPDATE rows (i.e., the delivered=true flip from Path A). Only INSERTs reach Kafka.
Skips sentinel rows where event_type is NULL.
Throws for malformed real events so Kafka Connect can retry, log, or skip according to your connector error settings.
Encodes the row as Avro bytes using ActionEvent.avsc plus the per-event_type payload schemas you mounted into the Connect container. No Schema Registry is required by the bundled SMT.

For Protobuf, swap OutboxToAvroTransform → OutboxToProtobufTransform and use descriptor-set paths instead of Avro schema paths. See Streaming (Debezium) for the full SMT options and the DLQ caveat for source connector failures.

B.2 Consume from Kafka

Any Kafka consumer in any language can subscribe. Example with the Avro consumer in Java:

var props = new Properties();
props.put("bootstrap.servers", "kafka:9092");
props.put("group.id", "warehouse");
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.ByteArrayDeserializer");

try (var consumer = new KafkaConsumer<String, byte[]>(props)) {
    consumer.subscribe(List.of("ekbatan.com.example.finance"));
    while (true) {
        for (var record : consumer.poll(Duration.ofSeconds(1))) {
            var bytes = record.value();
            // Decode bytes as ActionEvent, then decode ActionEvent.payload
            // with the schema matching ActionEvent.eventType.
        }
    }
}

The Avro and Protobuf integration tests include small consumers and routers that show the full decode path. They are test scaffolding, not a production consumer framework.

B.3 Fan out or rekey downstream

The connector does not have to publish directly to every final consumer topic. A common production shape is:

ekbatan.com.example.finance
        │
        ├─ Kafka Streams / ksqlDB / Flink / Pulsar Functions / custom router
        │
        ├─ ekbatan.com.example.finance.event.WalletMoneyDepositedEvent
        ├─ ekbatan.com.example.finance.model.Wallet
        └─ billing.wallet-deposits-by-owner

Keep the raw topic as the append-only stream of everything the service emitted. Then derive more specific topics from it. Derived topics can have different retention, schemas, partition counts, or message keys, without changing how actions write to the database.

Idempotency and ordering

All delivery paths are at-least-once. That is true for local handlers, handler-published broker messages, CDC connectors, and downstream routers. A process can crash after doing the side effect but before recording success; a broker publish can succeed and the publisher can still retry; a CDC connector can replay after a restart.

Design consumers so duplicate delivery is harmless:

Store the event row id, or a domain-specific idempotency key, in the consumer’s table before applying the side effect.
Put a unique constraint on that key so duplicate attempts become no-ops.
Use model_id as the broker message key when ordering matters per aggregate.
Keep handler names stable. If you rename a local handler, use aliases() until old event_notifications.handler_name rows drain.

What just happened

Local consumption keeps event delivery inside your application — same database, no broker, no external infrastructure to operate.
Handler publishing lets application code publish to one or many broker destinations with custom keys, headers, schemas, and routing.
CDC to a raw topic turns committed event rows into a cross-service event stream without publishing from request handlers.
CDC plus downstream routing lets one raw stream become many topics or differently keyed streams.
All of them read from the same eventlog.events rows. The dual-write trap is structurally impossible — there is only one durable write inside the action’s transaction, and everything else consumes from it.

If you start with local consumption and later add CDC, no application code changes. The Debezium connector starts tailing the same table from now on; backfill earlier events by configuring snapshot.mode.

→ Adding a shard — take the single-database wallet to two shards using EmbeddedBitsShardingStrategy. Your action signatures don’t change.