Consumers

A consumer reads topics and applies their messages: a projection to its read models, a route consumer by sending them to a destination. This page covers a consumer’s progress, what it guarantees, how instances share it and which states it is in.

Checkpoint and lag

A consumer keeps a checkpoint for every partition of every topic it reads: how far it has read there. It resumes after its checkpoints, on any instance and after any restart. An @InMemoryProjection is the exception: its checkpoints live in memory, so they belong to its one instance and end with it; see In-memory projections.

Its lag in a partition is the partition’s high-water mark minus its checkpoint, and its lag in a topic is that of its slowest partition. Lags of different topics are never added up, since the topics share no order.

On a topic with one position space, such as the event store, every partition reports the topic’s one high-water mark, and a lag counts positions rather than messages: the positions between a checkpoint and the mark hold other partitions' messages too.

A consumer whose writes commit with its checkpoint saves it with every message it applies; any other saves it once per page it reads. Over positions that hold nothing for a partition, either saves it at most once a minute, so a consumer that is caught up can show a lag of up to a minute of the topic’s other traffic.

GET /actuator/consumers/projection/account-summary shows a consumer’s checkpoint, high-water mark and lag per topic and partition; see the consumers endpoint. The topics endpoint shows each topic’s slowest consumers. A consumer’s report holds its state, its halt reason, its undelivered counts and any missing history; see the consumers endpoint.

Delivery guarantees

A consumer receives every message the topic still holds at least once; on a read-model topic, compaction can remove a superseded state before a slow consumer reads it, and the consumer then receives the latest state instead. It applies a message, then saves its checkpoint past it, so after a crash it receives again only what its checkpoint had not passed yet.

A consumer’s own writes are applied exactly once when they commit in the transaction that saves its checkpoint. The checkpoints live in the application’s primary datasource, so the consumer’s writes must go there, through the application’s transaction manager. For a projection its store decides this; see Two delivery paths. A route consumer sends to a system outside that transaction, so a crash between a send and its checkpoint sends the message again; see Route consumers.

A consumer whose writes commit on their own applies again, after a crash, at most the page it was in: up to 500 positions by default. Keep its effects idempotent.

Where a consumer’s writes commit with its checkpoint, an apply made by an instance that has just lost the partition to another instance rolls back, so each change still commits once across a handover.

Distribution

How the instances running a consumer share it is its distribution:

partitioned, the default: The consumer’s partitions are spread over the instances running it, and each applies its share, so adding an instance adds throughput. Messages of different partitions are applied in parallel, in no order relative to each other.
single-owner: One instance applies every partition, one message at a time, in position order across all of a topic’s partitions; the other instances stand by, ready to take over. Its throughput is one instance’s.
on every instance: An in-memory projection runs on every instance, each with its own read models, whatever its distribution says.

A projection’s distribution is spring.ddd.cqrs.projections.<name>.distribution (see Distribution); a route consumer’s is spring.ddd.routing.consumers.<destination>.distribution, else spring.ddd.routing.consumer.distribution. Switching a consumer between partitioned and single-owner keeps its progress.

Handover

Each partition of a consumer is applied by one instance at a time. When an instance joins, stops or crashes, only the partitions whose owner changes move; how long that takes, and what it costs, is described in When an instance joins or leaves.

A consumer’s report names each partition’s instance and host and its state: OWNED by a live instance, ORPHANED by an instance that is gone, or UNCLAIMED. GET /actuator/instances lists every instance with the partitions it holds; see the instances endpoint.

States

A consumer is in one of four states, the same on every instance that runs it, except an `@InMemoryProjection’s, which is its own instance’s alone:

RUNNING: It applies messages.
PAUSED: An operator paused it. It applies nothing until an operator resumes it.
HALTED: The framework stopped it on a problem, for a reason that decides how the halt ends; see Halts.
REBUILDING: A rebuild holds it, which only a projection has; see Rebuild and schema evolution.

See States on Running projections for what moves a consumer between them.

Pausing and resuming

Pausing, resuming and their refusals are described, for projections and route consumers alike, in Pausing and resuming.

Reading consumers from code

The status, pause, resume, resumePastGap and resumePastPoisonMessage methods of ConsumerLifecycle take the consumer’s key: its type and its name. A route consumer’s is the type route and its destination’s name. consumers() lists a report of every consumer the application runs, ordered by type and name, and report reads one by its key; both work inside a transaction. A high-water mark or a read-model count that cannot be read, from a database that cannot be reached, say, is reported as unknown and logged, so the rest of the report and every other consumer’s report stay readable. An in-memory projection reports the positions of the instance it is read on, and no ownership. A projection whose store can count its read models reports their number: the in-memory, JPA and query stores can, a projection writing its own tables cannot. A store in a database counts them each time a report is read, by scanning its table or an index. Read reports outside a transaction where you can. Inside one, a report’s reads of a topic or a read-model store on the transaction’s database join it: under REPEATABLE READ or a snapshot they see the topic as it was when the transaction began, while the checkpoints may be read on a connection of their own, so a lag can show too low. Each database a report reads also binds a connection to the transaction until it ends. A key the application does not run fails with NoSuchElementException. A consumer is listed once its projection or route has started, so a call before the application has started lists none.