Skip to content
OccurrentOccurrent

Durable subscriptions

Layer 0: CloudEvents · Blocking · Modules: occurrent-subscription-durable-blocking

Storing the checkpoint is useful if you need to resume a subscription from its last known checkpoint when restarting an application. Occurrent provides a utility that implements SubscriptionModel and combines a CheckpointAwareSubscriptionModel and a CheckpointStorage implementation (see here) to automatically store the checkpoint, by default,
after each processed event. If you don’t want the checkpoint to be persisted after every event, you can control how often this should happen by supplying a predicate to DurableSubscriptionModelConfig. There’s a pre-defined predicate, org.occurrent.subscription.util.predicate.EveryN, that allow
the checkpoint to be stored for every n event instead of simply every event. There’s also a shortcut, e.g. new DurableSubscriptionModelConfig(3) that creates an instance of EveryN that stores the checkpoint for every third event.

When the filter of a subscription matches no events for a long time, its stored checkpoint stays at the last event it handled. The MongoDB oplog can drop that position while the subscription is still up to date. A resume after a pause, a lease handover between competing consumers, or a restart would then start from a position the oplog no longer has.

To prevent that, DurableSubscriptionModel saves a quiet position when a read of the change stream returns no event for a subscription. That is how far the wrapped model, the CheckpointAwareSubscriptionModel it combines with the storage, has read, saved as the subscription’s checkpoint. It saves at most once a minute. The minute starts over whenever a checkpoint is saved for an event, so a subscription that stores a checkpoint for an event at least once a minute gets no extra write.

Only a wrapped model that implements QuietPositionReportingSubscriptions reports that position. NativeMongoSubscriptionModel and SpringMongoSubscriptionModel do.

Change the interval with saveQuietPositionEvery(Duration), and keep it well below how long the oplog keeps history:

var config = new DurableSubscriptionModelConfig(1).saveQuietPositionEvery(Duration.ofSeconds(30));

neverSaveQuietPosition() turns the save off.

A persist predicate that declines some events, such as EveryN with n above 1, can delay the save. A subscription that stops receiving events right after an event the predicate declined gets no position saved until the predicate stores one. A subscription from a StartAt of your own gets none saved until the predicate has stored the position of an event.

If you want full control, it’s recommended to pick a checkpoint storage implementation, and store the checkpoint yourself using its API.

To use it, first we need to add the dependency:

<dependency>
    <groupId>org.occurrent</groupId>
    <artifactId>occurrent-subscription-durable-blocking</artifactId>
    <version>0.34.0</version>
</dependency>

Then we should instantiate a CheckpointAwareSubscriptionModel, that subscribes to the events from the event store, and an instance of a CheckpointStorage, that stores the checkpoint, and combine them to a DurableSubscriptionModel:

// Create the non-durable blocking subscription instance
CheckpointAwareSubscriptionModel nonDurableSubscriptionModel = ...
// Create the storage
CheckpointStorage storage = ...


// Now combine the non-durable subscription model and the checkpoint storage
SubscriptionModel durableSubscriptionModel = new DurableSubscriptionModel(nonDurableSubscriptionModel, storage);


// Start a subscription
durableSubscriptionModel.subscribe("mySubscriptionId", cloudEvent -> doSomethingWithTheCloudEvent(cloudEvent));

When No Start Position Can Be Recorded

Section titled “When No Start Position Can Be Recorded”

A new subscription that asks for StartAt.subscriptionModelDefault() gets its start position stored before any event is delivered. DurableSubscriptionModel gets that position by calling globalCheckpoint() on the subscription model it wraps, inside subscribe(..). If the application crashes before the first checkpoint is saved, the subscription resumes from that stored position.

When globalCheckpoint() returns null and nothing is stored for the subscription, subscribe(..) throws IllegalStateException and registers nothing. NativeMongoSubscriptionModel and SpringMongoSubscriptionModel return null when the server refuses the hostInfo command, which a shared MongoDB Atlas cluster does.

A subscription that already has a checkpoint stored is never refused this way, and neither is one that passes a StartAt of its own.

To start such a subscription anyway, set startWhenNoStartPositionCanBeRecorded(true) on its config:

DurableSubscriptionModelConfig config = new DurableSubscriptionModelConfig(1).startWhenNoStartPositionCanBeRecorded(true);
DurableSubscriptionModel subscriptionModel = new DurableSubscriptionModel(wrappedSubscriptionModel, checkpointStorage, config);

The subscription then starts with no stored position. If the application crashes before the first checkpoint is saved, the subscription starts over from wherever the feed has got to by the time it restarts. An event whose delivery failed before the crash is then not delivered again.