Skip to content
OccurrentOccurrent

Blocking subscription models

Layer 0: CloudEvents · Blocking · Modules: occurrent-subscription-mongodb-native-blocking, occurrent-subscription-mongodb-spring-blocking, occurrent-subscription-inmemory

These are the non-durable blocking subscription implementations:

MongoDB

In-Memory

By “non-durable” we mean implementations that doesn’t store the checkpoint in a durable storage automatically.
It might be that the datastore does this automatically or that checkpoint storage is not required for your use case. If the datastore doesn’t support storing the checkpoint automatically, a subscription will typically implement the org.occurrent.subscription.api.blocking.CheckpointAwareSubscriptionModel interface (since these types of subscriptions needs to be aware of the checkpoint).

Typically, if you want the stream to continue where it left off on application restart you want to store away the checkpoint. You can do this anyway you like, but for most cases you probably want to look into implementations of org.occurrent.subscription.api.blocking.CheckpointAwareSubscriptionModel. These subscriptions can be combined with a checkpoint storage implementation to store the checkpoint in a durable datastore.

Occurrent provides a utility that combines a CheckpointAwareSubscriptionModel and a CheckpointStorage (see here) to automatically store the checkpoint
after each processed event. You can tweak how often the checkpoint should be persisted in the configuration.

Blocking Subscription using the “Native” Java MongoDB Driver

Section titled “Blocking Subscription using the “Native” Java MongoDB Driver”

Uses the vanilla Java MongoDB synchronous driver (no Spring dependency is required).

To get started first include the following dependency:

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

Then create a new instance of NativeMongoSubscriptionModel and start subscribing:

MongoDatabase database = mongoClient.getDatabase("some-database");
// Create the blocking subscription
SubscriptionModel subscriptionModel = new NativeMongoSubscriptionModel(database, "eventCollection", TimeRepresentation.DATE, Executors.newCachedThreadPool(), RetryStrategy.retry().fixed(200));


// Now you can create subscription instances that subscribes to new events as they're written to an EventStore
subscriptionModel.subscribe("mySubscriptionId", cloudEvent -> doSomethingWithTheCloudEvent(cloudEvent));


// You can later cancel the subscription by calling:
subscriptionModel.cancelSubscription("mySubscriptionId");

There are a few things to note here that needs explaining. First we have the TimeRepresentation.DATE that is passed as the third constructor argument which you can read more about here. Secondly we have the Executors.newCachedThreadPool(). A thread will be created from this executor for each call to “subscribe” (i.e. for each subscription). Make sure that you have enough threads to cover all your subscriptions or the “SubscriptionModel” may hang. Last we have the RetryStrategy which defines what should happen if there’s e.g. a connection issue during the life-time of a subscription or if subscription fails to process a cloud event (i.e. the action throws an exception).

Note that you can provide a filter, start position and checkpoint persistence for this subscription implementation.

Blocking Subscription using Spring MongoTemplate

Section titled “Blocking Subscription using Spring MongoTemplate”

An implementation that uses Spring’s MongoTemplate for event subscriptions.

First include the following dependency:

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

Then create a new instance of SpringMongoSubscriptionModel and start subscribing:

MongoTemplate mongoTemplate = ...
// Create the blocking subscription
SubscriptionModel subscriptionModel = new SpringMongoSubscriptionModel(mongoTemplate, "eventCollectionName", TimeRepresentation.RFC_3339_STRING);


// Now you can create subscriptions instances that subscribes to new events as they're written to an EventStore
// Typically you do this after the Spring application context has finished loading. For example by subscribing to
// the  (org.springframework.boot.context.event.ApplicationStartedEvent) or in a method annotated with (@PostConstruct)
subscriptionModel.subscribe("mySubscriptionId", cloudEvent -> doSomethingWithTheCloudEvent(cloudEvent));


// You can later cancel the subscription by calling:
subscriptionModel.cancelSubscription("mySubscriptionId");

The “eventCollectionName” specifies the event collection in MongoDB where events are stored. It’s important that this collection is the same as the collection used by the EventStore implementation. Secondly, we have the TimeRepresentation.RFC_3339_STRING that is passed as the third constructor argument, which you can read more about here. It’s also very important that this is configured the same way as the EventStore.

Each subscription reads its change stream with a cursor of its own, on a thread from the model’s executor. When the change stream fails, for example because the connection to MongoDB is lost, the model opens it again from the position the subscription had reached, and retries that with its RetryStrategy.

When it comes to retries, if the “action” fails (i.e. if the higher-order function you provide when calling subscribe throws an exception), either using something like Spring Retry or the Occurrent Retry Module. By default, all subscription models will use the Occurrent retry module with exponential backoff starting with 100 ms and progressively go up to max 2 seconds wait time between each retry when reading/saving/deleting the checkpoint. You can customize this by passing an instance of RetryStrategy to the SpringMongoSubscriptionModel constructor.

When the action still throws after the RetryStrategy gives up, the model opens the change stream again at the position before that event and delivers the event again, so the event isn’t skipped.

The same RetryStrategy decides how many times the model opens the change stream again. It counts every reopen since the subscription was started or resumed, including those after a lost connection. When it gives up, the model logs an error, and the subscription receives nothing more until you pause and resume it. With RetryStrategy.none() that happens the first time the action throws.

If you want to disable the Occurrent retry module, pass RetryStrategy.none() to the SpringMongoSubscriptionModel constructor and then handle retries anyway you find fit. For example, let’s say you want to use spring-retry, and you have a simple Spring bean that writes each cloud event to a repository:

@Component
public class WriteToRepository {


  private final SomeRepository someRepository;


  public WriteToRepository(SomeRepository someRepository) {
    this.someRepository = someRepository;
  }


  public void write(CloudEvent cloudEvent) {
    someRepository.persist(cloudEvent);
  }
}

And you want to subscribe to all “GameStarted” events and write them to the repository:

WriteToRepository writeToRepository = ...
subscriptionModel.subscribe("gameStartedLog", writeToRepository::write);

But if the connection to someRepository is flaky you can add Spring Retry so allow for retry with exponential backoff:

@Component
public class WriteToRepository {


  private final SomeRepository someRepository;


  public WriteToRepository(SomeRepository someRepository) {
    this.someRepository = someRepository;
  }


    @Retryable(backoff = @Backoff(delay = 200, multiplier = 2, maxDelay = 30000))
  public void write(CloudEvent cloudEvent) {
    someRepository.persist(cloudEvent);
  }
}

Don’t forget to add @EnableRetry in to your Spring Boot application as well.

Note that you can provide a filter, start position and checkpoint persistence for this subscription implementation.

Restart Subscription when Oplog Lost

Section titled “Restart Subscription when Oplog Lost”

From 0.34.0, a subscription whose filter matches no events for a long time no longer needs automatic restart. When a read of the change stream returns no event for it, SpringMongoSubscriptionModel moves its position forward anyway, so a resume after a pause opens at a position the oplog still has. A DurableSubscriptionModel wrapping the model also saves that position as the checkpoint once a minute, so a restart or a lease moving to another node resumes from it too.

You still need automatic restart when:

  • the subscription was paused, the application was down, or no node held the subscription’s lease, for longer than the oplog keeps history
  • no DurableSubscriptionModel wraps the model, and the subscription restarts from a position you stored yourself
  • the save is off (neverSaveQuietPosition()), or its interval (saveQuietPositionEvery(Duration)) is longer than the oplog keeps history
  • the persist predicate declined the last event before the subscription went quiet, as EveryN with n above 1 can, or the subscription started from a StartAt of your own and the predicate hasn’t stored a position yet

When the oplog no longer has the history a subscription needs to resume, the subscription model can restart the subscription from the current time instead. This section uses SpringMongoSubscriptionModel as the example, but NativeMongoSubscriptionModel and the reactive ReactorMongoSubscriptionModel work the same way, with the same restartSubscriptionsOnChangeStreamHistoryLost(true) setting on their config.

A subscription runs into this whenever it opens its change stream at a position the oplog no longer has. That happens when an application restarts with a subscription configured to start from such a position. It also happens when a subscription is resumed, or its change stream restarts, after being paused or disconnected for longer than the oplog keeps history. A subscription that hasn’t handled an event yet gets the same handling.

Automatic restart is off when you create the model yourself, because a subscription that restarts from the current time can miss events. That happens when the subscriptions run in a different process than the writes, and the event store gets many writes. It’s safe when the subscription runs in the same process as the writes, as long as the subscription starts before the application accepts writes.

To turn it on when you create the model yourself:

var subscriptionModel = new SpringMongoSubscriptionModel(mongoTemplate, SpringMongoSubscriptionModelConfig.withConfig("events", TimeRepresentation.RFC_3339_STRING).restartSubscriptionsOnChangeStreamHistoryLost(true));

With automatic restart turned off, the model logs an error and removes the subscription, so its id can be subscribed again.

With automatic restart turned on and a DurableSubscriptionModel wrapping the model, the position the subscription restarts from is stored as its checkpoint before the restart, unless another node has written the checkpoint with a newer lease. So a process restart before the next event doesn’t open the change stream at the lost position again.

When the lost position is the live start a catch-up subscription handed over, the model asks the catch-up first, whichever way automatic restart is set. The catch-up takes the subscription over, and the model pauses it instead of restarting or removing it.

An alternative approach to restarting automatically is to use a catch-up subscription and restart the subscription from an earlier date.

Deferring Startup (Added in 0.32.0)

Section titled “Deferring Startup”

By default a SpringMongoSubscriptionModel starts itself as soon as it’s constructed, and if you register it as a Spring bean, Spring starts it again on refresh (it implements SmartLifecycle). Set autoStartup(false) on the config to skip both:

var config = SpringMongoSubscriptionModelConfig
        .withConfig("events", TimeRepresentation.RFC_3339_STRING)
        .autoStartup(false);


var subscriptionModel = new SpringMongoSubscriptionModel(mongoTemplate, config);

A subscription registered on a model created this way comes back paused, the same as if you’d called stop() right after construction. No change stream opens until you call start() or resumeSubscription(id) yourself.

subscribe(..) asks MongoDB for its current operation time on the model’s executor, without waiting for the answer. A subscription from StartAt.now(), or with no StartAt, starts at that time. So it receives the events written from shortly after subscribe(..) returns, including those written before start().

Use autoStartup(false) to bring subscriptions up under your own control, behind a leader election or a health check for example, or in a test that wants to choose which subscriptions actually run before the context finishes starting.

Tuning the MongoDB change stream (Added in 0.31.0)

Section titled “Tuning the MongoDB change stream”

Both blocking MongoDB subscription models let you tune the change stream they read from. Both options are opt-in. Leave them unset and you get the driver and server defaults, which is what every subscription did before these existed, so upgrading changes nothing on its own.

batchSize sets how many change-stream documents the server returns per batch. A larger batch means fewer round trips to the server, which helps throughput on a high-volume subscription such as an outbox, and costs more memory per batch. Values in the low hundreds work well for high throughput, but the right number depends on your event size and load.

maxAwaitTime bounds how long the server holds an idle cursor before returning a batch, possibly an empty one. A shorter wait delivers events sooner and costs more getMore round trips while the stream is quiet. A longer wait keeps an idle cursor waiting and reduces that chatter. Somewhere between 200 ms and 1000 ms suits most workloads.

var config = NativeMongoSubscriptionModelConfig.withConfig()
        .batchSize(500)
        .maxAwaitTime(Duration.ofMillis(500));


var subscriptionModel = new NativeMongoSubscriptionModel(database, "events", TimeRepresentation.DATE, executor, config);

The two models don’t offer the same options. NativeMongoSubscriptionModel has both. SpringMongoSubscriptionModelConfig has maxAwaitTime only:

var config = SpringMongoSubscriptionModelConfig
        .withConfig("events", TimeRepresentation.RFC_3339_STRING)
        .maxAwaitTime(Duration.ofMillis(500));

Use NativeMongoSubscriptionModel when you need batchSize.

ReactorMongoSubscriptionModel offers neither.

InMemory Subscription

Section titled “InMemory Subscription”

If you’re using the InMemory EventStore you can use the “InMemorySubscriptionModel” to subscribe to new events. For add the dependency:

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

Then you can use it like this:

InMemorySubscriptionModel subscriptionModel = new InMemorySubscriptionModel();
InMemoryEventStore inMemoryEventStore = new InMemoryEventStore(inMemorySubscriptionModel);


subscriptionModel.subscribe("subscription1", System.out::println);


// You can later cancel the subscription by calling:
subscriptionModel.cancelSubscription("mySubscriptionId");

By default this model refuses a subscription filter on the event data payload. Pass a DataFieldReader to its constructor to give it one, see Filtering on Payload Data.