Bookmarking

Stream Readers and Bookmarking

This guide covers how to track event processing progress using bookmarks, enabling resumable and eventually consistent event processing.

Stream Readers and Bookmarking

In event sourcing, it’s crucial to persist the point up to which a stream has been processed by a reader. This enables:

• Resumable processing: Restart processing from the last checkpoint after crashes or deployments
• Eventually consistent read models: Build derived views that catch up independently
• Processing guarantees: Ensure events are processed at-least-once
• Progress monitoring: Track how far behind real-time each reader is

Eventual consistency allows readers and processors to move independently from the write side. A new read model that still has to catch up with history can process events at its own pace:

// New projection starting from scratch
CustomerSummary projection = new CustomerSummary("123");

// Process all historical events
Projector.from(stream).into(projection).build().run();

// Now caught up with history

Bookmarks enable this independence by marking the last successfully processed event — and, beside it, how far the reader has read the stream (see Two Positions: Handled and Read).

Placing a Bookmark

The placeBookmark() method registers the EventReference of the last processed event for a named reader:

stream.placeBookmark(
    "customer-summary-builder",        // Reader identifier
    event.reference(),                  // Last processed event
    Tags.of("status", "active")        // Metadata tags
);

Reader Identifier: A unique string identifying the processor (e.g., “order-projection”, “email-sender-v2”). This is the key for the bookmark.

Event Reference: The reference of the last event successfully processed by this reader.

Tags: Optional metadata stored with the bookmark for observability. Common uses:

• Processing status: Tags.of("status", "active")
• Timestamp: Tags.of("updated-at", Instant.now().toString())
• Version: Tags.of("version", "2.0")
• Instance ID: Tags.of("instance", hostname)

Tags are stored only for monitoring and debugging—they don’t affect bookmark retrieval.

Creating vs. Updating: There’s no distinction. Calling placeBookmark() with the same reader ID creates the bookmark on first call and updates it on subsequent calls. The reader ID is the key.

// First call - creates bookmark
stream.placeBookmark("my-reader", ref1, Tags.none());

// Second call - updates bookmark for same reader
stream.placeBookmark("my-reader", ref2, Tags.none());

The Reference Must Name a Stored Event

A bookmark is a position in this store’s log, so the reference has to name an event this store actually holds. One it has never seen — in practice a reference carried over from a different store or prefix — is a caller error: placeBookmark throws an EventStorageException, nothing is written, and any bookmark the reader already had stays exactly as it was, reference and tags included.

// a reference obtained from another store, or fabricated in a test
stream.placeBookmark("my-reader", referenceFromSomewhereElse, Tags.none());
// -> EventStorageException; "my-reader" still points where it did before

The same holds for a bookmark’s read position (below). This is the contract on every backend — PostgreSQL enforces it with the fk_bookmarks_event_id and fk_bookmarks_read_up_to_event_id foreign keys, the in-memory backends check their own log — and the TCK pins it per backend. Without it, a miswired multi-store setup poisons a reader’s cursor silently, and the reader then resumes from a position that means nothing in the store it is reading.

A Bookmark Stores the Event Id, and Nothing Else About the Event

A bookmark records which event the reader reached, by id — and, when it has one, its read position, by id too. Its position and transaction — the coordinates cursor comparisons order by — are answered from the stored event itself whenever the bookmark is read: on PostgreSQL by joining the events row on its unique id, in memory by resolving the reference from the log.

So the reference you read back — and the one a bookmark notification carries, read position included — is always the store’s own for that event, whatever the caller passed in. Two things follow:

• A bookmark cannot carry a cursor that disagrees with the event it names. Storing the caller’s (tx, position) beside the id would let a bookmark pass validation with a stored id and a wrong cursor, since the check is on the id.
• A bookmarks table is valid in another store holding the same events. An import preserves event ids and reassigns both ordering columns, so bookmarks copied across by id resolve to the target’s own coordinates as they stand.

Two Positions: Handled and Read

A bookmark holds two positions, because they answer two different questions:

• reference — the last event the reader handled. This is where the reader resumes.
• readUpTo — the event up to which the reader has read the stream, relevant to it or not. This is what its backlog is counted from.

They differ for any reader whose query names only some event types. A projection over OrderPlaced and OrderShipped never handles a PaymentReceived, so its reference never moves past one. Counting “events after the reference” as its backlog therefore counts every payment appended since its last order event — events it will never handle, for good. A reader that is fully caught up would look further behind with every payment. readUpTo moves past those events, so the backlog counted from it is only what the reader has not read yet.

stream.placeBookmark(
    "order-dashboard",
    lastHandled.reference(),            // where the reader resumes
    headWhenTheRunStarted,              // how far it has read, handled or not
    Tags.none()
);

A bookmarked Projector places both positions itself, so code bookmarking by hand rarely needs this four-argument overload. The three-argument placeBookmark(reader, reference, tags) records no read position.

The read position is never a resume point. Resuming from it would buy nothing — the typed query already skips the events the reader does not handle, through the index — and it would lose events silently: a reader whose query later gains an event type would never be handed the events of that type sitting between the two positions.

Either position may be absent, but not both. A bookmark always says something:

• No read position — a bookmark placed without one reads back with an empty readUpTo, and Bookmark.readUpToOrReference() falls back to the reference, which is exactly what such a bookmark says.
• No handled event — “read up to here, handled nothing yet”. A reader whose query selects event types that have not occurred yet reads the stream without handling anything. Without a bookmark its whole stream would count as backlog, although nothing in it concerns it. So it records its read position alone, and reference() reads back empty.

stream.placeReadPosition(
    "order-dashboard",
    headWhenTheRunStarted,              // how far it has read; it has handled nothing yet
    Tags.none()
);

A reader with such a bookmark still resumes from the beginning, exactly as one without a bookmark does: getBookmark(reader) — the resume point — answers empty, since the read position is never a resume point. A Projector built with resumeAfterReadPosition() is the deliberate exception, for a reader started from now on. Placing a read position alone never clears a handled reference the bookmark already names; only the read position, tags and update time are replaced. The first placeBookmark that names a handled event fills the reference in. A bookmarked Projector places a read-position-only bookmark itself, after a run that read to the end without handling anything.

It is held to the same rules as the reference, independently: it must name an event this store holds (or the placement is rejected and the previous bookmark stays), only its event id is stored, and it reads back as the store’s own coordinates for that event — as the two sections above describe. A storage does not judge whether readUpTo is at or after reference; that is the writer’s to keep.

Retrieving a Bookmark

The getBookmark() method retrieves the last bookmarked position for a reader:

Optional<EventReference> bookmark = stream.getBookmark("customer-summary-builder");

bookmark.ifPresentOrElse(
    ref -> System.out.println("Resume from position: " + ref.position()),
    () -> System.out.println("No bookmark - start from beginning")
);

Returns Optional.empty() if no bookmark exists for that reader (first run scenario), or if its bookmark records a read position only — a reader that has handled nothing yet starts from the beginning either way.

getBookmark() answers the resume point only. findBookmark() answers the whole bookmark of one reader — both positions, its tags, and when it was last placed — including the bookmark of a reader that has read the stream without handling anything yet:

Optional<Bookmark> bookmark = stream.findBookmark("order-dashboard");

bookmark.ifPresent(b -> System.out.printf(
    "handled up to %s, read up to %d%n",
    b.reference().map(r -> String.valueOf(r.position())).orElse("nothing yet"),
    b.readUpToOrReference().position()));

It is the read for a caller that wants to know how far a reader has read — for instance one waiting for a reader to have seen a given event:

// has the reader read past the event I just appended?
EventReference appended = stream.append(event).getLast().reference();

boolean seen = stream.findBookmark("order-dashboard")
    .map(b -> !appended.happenedAfter(b.readUpToOrReference()))
    .orElse(false);

Comparing with the read position rather than the reference is what makes this work for an event the reader does not handle: its reference never reaches such an event, while its read position does.

Retrieving All Bookmarks

getBookmarks() returns every bookmark on the stream, with the metadata that getBookmark() does not carry:

public record Bookmark (
    String reader,                     // the reader identifier
    Optional<EventReference> reference, // the last event it handled: where it resumes; empty when it has handled nothing yet
    Optional<EventReference> readUpTo,  // how far it has read the stream; empty when none was recorded
    Tags tags,                          // the tags supplied when it was placed
    Instant updatedAt                   // when it was last placed
) {
    public EventReference readUpToOrReference() { ... } // readUpTo when recorded, reference otherwise; never null
}

At least one of the two positions is always present — the record refuses a bookmark naming neither — so readUpToOrReference() always answers.

List<Bookmark> bookmarks = stream.getBookmarks();

bookmarks.forEach(b -> System.out.printf(
    "%-30s handled %-8s read %-8d updated %s %s%n",
    b.reader(),
    b.reference().map(r -> String.valueOf(r.position())).orElse("-"),
    b.readUpToOrReference().position(), b.updatedAt(), b.tags()));

This is the API behind an operational view of a system’s readers. Where getBookmark(reader) answers “where is this processor”, getBookmarks() answers “where is everything, and which of it has stopped moving”:

// which readers have fallen behind the head of the stream?
Optional<EventReference> head = stream.head();

stream.getBookmarks().stream()
    .filter(b -> head.isPresent() && b.readUpToOrReference().happenedBefore(head.get()))
    .forEach(b -> LOGGER.warn("reader {} is behind, last moved at {}", b.reader(), b.updatedAt()));

// which readers have not moved in an hour?
Instant cutoff = Instant.now().minus(Duration.ofHours(1));
stream.getBookmarks().stream()
    .filter(b -> b.updatedAt().isBefore(cutoff))
    .forEach(b -> alertOnStalledReader(b.reader(), b.updatedAt()));

Lag is measured from readUpToOrReference(), never from reference(): a reader whose query skips most event types would otherwise always look behind, however current it is. Note that a bookmarked Projector with nothing to handle moves its read position at most once per idle bookmark interval (two seconds by default), so allow that much slack before calling a reader behind.

updatedAt is stamped at placement time, and tags are exactly the tags passed to placeBookmark(...) — which is what makes them worth populating with a hostname, an instance id or an application version. A reader that has stopped moving is much easier to chase down when its bookmark says which process last touched it.

Removing a Bookmark

removeBookmark(reader) deletes a reader’s bookmark and returns the handled reference it held, or Optional.empty() if there was none — or if it recorded a read position only, which it removes too:

Optional<EventReference> removed = stream.removeBookmark("customer-analytics");

The usual reason to do this is a projection rebuild: dropping the read model and the bookmark together makes the next run replay from the beginning of the stream.

readModel.truncate();
stream.removeBookmark("customer-analytics");
projector.run();                    // replays from the start

Typical Usage Scenario

A complete example showing bookmark retrieval, event processing with append listeners, and periodic bookmark updates:

EventStream<CustomerEvent> stream = eventstore.getEventStream(
    EventStreamId.forContext("customer").anyPurpose(),
    CustomerEvent.class
);

String readerName = "customer-analytics";

// Retrieve last bookmark
Optional<EventReference> bookmark = stream.getBookmark(readerName);
EventReference startAfter = bookmark.orElse(null);

if (startAfter == null) {
    System.out.println("No bookmark found - processing from beginning");
} else {
    System.out.println("Resuming from position: " + startAfter.position());
}

// Register listener for new events appended to stream
Subscription subscription = stream.subscribe((EventReference atLeastUntil) -> {
    // Process new events since the last bookmark
    EventReference from = stream.getBookmark(readerName).orElse(null);
    for (Event<CustomerEvent> event : stream.query(EventQuery.matchAll().limit(100), from)) {
        processEvent(event);

        // Update bookmark after processing
        stream.placeBookmark(
            readerName,
            event.reference(),
            Tags.of("processed-at", Instant.now().toString())
        );
    }
    return null;
});

// Initial catch-up: process all existing events
for (Event<CustomerEvent> event : stream.query(EventQuery.matchAll(), startAfter)) {
    processEvent(event);
    stream.placeBookmark(readerName, event.reference(), Tags.none());
}

When the application starts:

• No bookmark exists: Process from the beginning of the stream
• Bookmark exists: Resume from the bookmarked position

The append listener ensures new events are processed without delay.

This simple/naive implementation does one write operation (placeBookmark) per single event that is processed, which is far from ideal.
The goal here was to show the basic setup, read further for ways to optimize this.

Typical Usage Scenario with Batching

A single reader instance works through an event stream, placing bookmarks as savepoints:

String readerName = "order-fulfillment";
EventStream<OrderEvent> stream = eventstore.getEventStream(
    EventStreamId.forContext("orders").anyPurpose(),
    OrderEvent.class
);

// Retrieve bookmark
Optional<EventReference> lastProcessed = stream.getBookmark(readerName);

// Query events after bookmark, 500 stored events at a time
EventQuery query = EventQuery.matchAll().limit(500);
EventReference after = lastProcessed.orElse(null);

while (true) {
    EventPage<OrderEvent> page = stream.page(query, after);

    if (page.isExhausted()) {
        break; // Caught up
    }

    // Process batch
    page.events().forEach(event -> processEvent(event));

    // Update bookmark after batch: the last stored event read, even if it upcast into nothing
    EventReference lastInBatch = page.lastStoredEventReference().orElseThrow();
    stream.placeBookmark(
        readerName,
        lastInBatch,
        Tags.of("batch-size", String.valueOf(page.events().size()))
    );

    after = lastInBatch;
}

This hand-written loop is what a bookmarked Projector does for you — Projector.from(stream).into(projection).bookmarkAs(readerName).build().run() — including rolling its cursor back when a batch fails.

Restart Behavior: If the process crashes and restarts, the bookmark is a persisted pointer indicating where to resume. The reader queries events after the bookmark and continues processing.

Integration with Query Parameters: The bookmark works naturally with the after parameter of event queries. Simply pass the bookmarked reference as the after parameter.

Multi-Server Failover: In a multi-server setup with failover, another logical instance of the same reader can take over by retrieving the same bookmark:

// Server 1 goes down while processing
// Bookmark persisted at position 1000

// Server 2 takes over
Optional<EventReference> bookmark = stream.getBookmark("order-fulfillment");
// Resumes from position 1000

Multiple instances can’t process the same reader concurrently (sequential processing), but failover enables high availability.

A bookmark says where a reader got to; it does not say which instance is entitled to move it. Deciding that is what Leader Election with Leases is for: the standby instances hold the same reader name and take it over only once the lease says they own it.

Idempotent Processing with Larger Batches: When combined with idempotent event handling, bookmarks can be updated less frequently for better performance:

int bookmarkEvery = 100; // Update bookmark every 100 events
int processedCount = 0;
EventReference lastEventRef = null;

for (Event<OrderEvent> event : stream.query(EventQuery.matchAll().limit(1000), after)) {
    processEventIdempotently(event); // Idempotent handler
    processedCount++;
    lastEventRef = event.reference();

    if (processedCount % bookmarkEvery == 0) {
        stream.placeBookmark(readerName, lastEventRef, Tags.none());
    }
}

// Place final bookmark
if (lastEventRef != null) {
    stream.placeBookmark(readerName, lastEventRef, Tags.none());
}

If processing fails mid-batch, idempotent handlers allow re-processing without side effects.

Idempotency can be as simple as storing the Event position (available in the EventReference) along with the read model, and to ignore handling of any events that carry a position that is not higher than the one already stored.

Multiple Reader Instances

Event streams are typically processed sequentially because events in a stream have causal relationships and ordering requirements. Parallelizing their processing toward a useful result is not trivial.

Exceptions: Scenarios that can be parallelized include:

• MapReduce algorithms: Independent aggregations that can be combined
• CRDTs (Conflict-free Replicated Data Types): Structures designed for commutative operations

For most use cases, a single reader instance works its way sequentially through all events in a stream.

However, multiple independent read models can be built in parallel:

// Three different readers processing the same stream independently

// Reader 1: Customer analytics
stream.placeBookmark("customer-analytics", ref1, Tags.none());

// Reader 2: Email notification sender
stream.placeBookmark("email-sender", ref2, Tags.none());

// Reader 3: Reporting dashboard
stream.placeBookmark("dashboard-updater", ref3, Tags.none());

Each reader:

• Processes events at its own pace
• Maintains its own bookmark
• Can lag behind or catch up independently
• Builds a different view or performs different actions

These are the “readers” placing their bookmarks. They operate independently on the same event stream.

Key Insight: While a single reader must process events sequentially, many different readers can process the same stream concurrently, each maintaining their own bookmark.

Listening for Bookmark Updates - Monitoring Reader Progress

Register a listener to receive real-time notifications when bookmarks are updated:

Subscription subscription = stream.subscribe((String reader, EventReference processedUntil) -> {
    System.out.println(
        reader + " processed up to position " + processedUntil.position()
    );

    // Record metrics
    metricsService.recordReaderPosition(reader, processedUntil.position());
});

The listener — a BookmarkListener — is invoked asynchronously (eventually consistent) whenever any reader places or updates a bookmark. The reference it receives is the bookmark’s reference — the last event the reader handled — with the store’s own coordinates for that event. Where the read position matters, as it does for lag or for waiting on a reader, look it up with stream.findBookmark(reader) when notified. Close the returned Subscription to stop listening.

A placement that records a read position alone — a reader that has read the stream without handling anything yet — has no processed-until to report, so it is not passed to bookmarkUpdated. It goes to BookmarkListener.readPositionUpdated(reader, readUpTo), a default method that does nothing. A listener waiting for a reader to have read further — an automation waiting for its todo list, say — overrides it too:

stream.subscribe(new BookmarkListener() {
    @Override
    public void bookmarkUpdated(String reader, EventReference processedUntil) {
        progress.put(reader, processedUntil);
    }

    @Override
    public void readPositionUpdated(String reader, EventReference readUpTo) {
        progress.put(reader, readUpTo);   // read this far, handled nothing yet
    }
});

Some use Cases for Bookmark Listeners:

• Observability - verify processors are still running and Eventual Consistency is keeping up with events being appended
• Monitoring delays in processing
• Waiting for a Reader to process up to a certain point
• Coordinating Multiple Readers
• etc …

Bookmark listeners enable powerful monitoring and coordination patterns for eventually consistent event processing systems. For a point-in-time view of every reader rather than a notification per change, use getBookmarks() as shown above.