Occurrent 0.33.0 is released. Where 0.32.0 made the event store and subscription contracts explicit, this release builds on top of them. The read side gets faster catch-up replays, a competing consumer that loses its lease can no longer overwrite the checkpoint left by the node that took over, and a flow saga step can wait for one of several alternatives instead of a single event. It builds on 0.32.0 and requires Java 21.

Highlights

Replay batching for materialized views

A materialized view built with Projections.materializedView(...) (or its reactor twin) used to read the store once and write it once for every event a catch-up replay fed it. CatchupProjectionFeed and DomainEventFeed now buffer replayed events per view instance and flush them in batches instead, controlled by MaterializedViewOptions’s batchSize (1000 by default). This is on by default, nothing to opt in to.

The shipped Mongo-backed ViewStateRepository implementations override findAllById and saveAll, so a flush is one bulk round trip each way instead of one call per view instance. A view you build by hand gets the same batching by implementing the new ReplayAware capability interface. See replay batching for what a replay does and does not promise about the view while it runs.

Checkpoint fencing for competing consumers

A node that lost its competing-consumer lock could still write a checkpoint after another node had already taken over and written a later one. That moved the checkpoint backward, and the new holder redelivered events it had already processed. CompetingConsumerStrategy can now hand a subscription model a fencing token, a number that only increases as the lock changes hands, and CheckpointStorage.save takes a CheckpointWriteCondition a subscription model uses to refuse a checkpoint write carrying an older token than the one already stored.

Register a CompetingConsumerStrategy bean in a Spring Boot application, and the starter wires the fence in automatically. If you wire your own subscription models, pass strategy::fencingToken as an extra constructor argument to whichever one writes the checkpoints, DurableSubscriptionModel and the catch-up models all take it. See checkpoint fencing for the full example.

A refused write throws CheckpointWriteConditionNotFulfilledException rather than moving the checkpoint, and neither Mongo subscription model retries it, since a lease that has already moved on never succeeds on a later attempt. The node’s own lease refresh notices within one lease period that it no longer holds the lock and pauses the consumer there, so a losing node stops instead of quietly overwriting the winner’s progress. This does not change the at-least-once contract this library has always kept, and it is not exactly-once delivery.

This breaks every CheckpointStorage implementation outside this repository, on both the blocking and reactor APIs, since save gains a condition argument and a new writeVersion method. If you implement in Java, org.occurrent.UpgradeToOccurrent_0_33 adds both methods for you. save delegates any() to your existing write and refuses anything stronger, and writeVersion answers empty, the correct permanent answer for a store that can’t evaluate a condition. The recipe is Java only, so a Kotlin implementer adds the same two members by hand. See the upgrade guide.

Step conditions for flow sagas

A flow saga step could branch on a single event with on, or wait for a list of events with join, but it could not wait for one of several alternatives. on now takes a condition as well as an event type, so on(anyOf(event<Approved>(2), event<Rejected>())) waits for either two approvals or a single rejection and reacts to whichever arrives first.

Conditions are built from event(type, count) checks combined with allOf(...) and anyOf(...). You can put one condition inside another, so a single step can ask for a count and an alternative at the same time. allOf(event<ItemPacked>(2), anyOf(event<CourierAssigned>(), event<PickupScheduled>())) waits for two packed items plus either a courier or a pickup slot, whichever of the two turns up. A condition is a plain value you can reuse across steps, and it goes alongside classic on(Class, ...) branches in the same step. Kotlin gets event<T>, allOf, anyOf and on(condition, then) to match.

join and Kotlin’s expect<T> are deprecated in favor of on(allOf(...)). Both keep working, so nothing forces a migration, but a lowered join’s reaction now reads only the events that arrived since its own step was entered, rather than everything the saga instance has received. See step conditions and ADR 120.

Smaller additions

  • ReactorCheckpointStorage now retries a transient MongoDB error while reading, saving or deleting a checkpoint, with the same exponential backoff SpringMongoCheckpointStorage already uses on the blocking stack, 100 ms up to 2 seconds, bounded to 5 attempts.
  • @EnableOccurrentTesting clears checkpoints on its own once the test context has exactly one CheckpointStorage bean, and a new clearState = true attribute does the same for the database flush once a store integration is available to flush with.
  • A filter on a data path through an array of objects could match the wrong events, picking up a same-named field nested inside a later sibling. It now matches only the path you asked for. A composed AND filter over several data paths also reads a byte-backed payload once rather than once per path.
  • A flow saga reaction can now ask whether an event type arrived, received.none(Rejected.class) or received.any(Rejected.class), rather than fetching the whole list and testing it for emptiness. Kotlin has reified none<T>() and any<T>().
  • A test names a flow step’s timer with stepTimer("awaiting-players") and hands it to SagaInput.timeout(sagaId, timerName) to fire that step’s timeout, or to SagaEffect.cancelTimeout(timerName) to assert on it. A timer name is a TimerName rather than a String, which breaks new SagaTimeout(sagaId, name) and reading timerName() into a String, but startTimeout, cancelTimeout, evolveOnTimeout and reactOnTimeout keep their string forms, so most saga code compiles unchanged. Kotlin has a top-level stepTimer and a TimerName overload everywhere the Java API gained one. See section 7 of the upgrade guide.

Renamed subscription capability interfaces

Five subscription capability interfaces are renamed, since none of them ever extended SubscriptionModel and the old names claimed a relationship they didn’t have. ReplayAwareSubscriptionModel and IntrospectableSubscriptionModel become ReplayAwareSubscriptions and IntrospectableSubscriptions on both the blocking and reactor stacks, and DelegatingSubscriptionModel becomes SubscriptionModelWrapper. SubscriptionModelWrapper’s two methods move with it, getDelegatedSubscriptionModel to getWrappedSubscriptionModel and getDelegatedSubscriptionModelRecursively to getWrappedSubscriptionModelRecursively. The published TCK base class IntrospectableSubscriptionModelConformance moves too, to IntrospectableSubscriptionsConformance.

Every capability interface now extends a new marker interface, SubscriptionModelCapability, and the static lookup each one carries is called findIn rather than of. of is the Java convention for building a value that never comes back empty, and this method searches a wrapper chain and can, so the old name promised something it did not deliver. SubscriptionModelCapability also adds capability(Class) and hasCapability(Class), so a caller already holding a capability’s Class can ask for it directly instead of calling the matching static findIn.

These, the saga timer name, and the CheckpointStorage change above are the only source-breaking changes in this release. org.occurrent.UpgradeToOccurrent_0_33 renames all five interfaces, SubscriptionModelWrapper’s two methods, the TCK base class, and both of methods for you, in Java and Kotlin alike. See the upgrade guide.

Behavior changes

These changes alter what already-running code does, with no compiler error to point at them.

  • CompetingConsumerSubscriptionModel.pauseSubscription(id) now actually pauses a consumer that has not won the lock yet. Before this fix it logged the call and returned as if it had succeeded, isPaused kept answering false, and the consumer started anyway the moment the lock arrived. Pausing now unregisters it from the strategy, so the lock never arrives while it is paused.
  • A MongoDB lease strategy (NativeMongoLeaseCompetingConsumerStrategy, SpringMongoLeaseCompetingConsumerStrategy) no longer races its scheduled lease refresh against an application thread registering or unregistering the same consumer, a race that could leave a lease held and refreshed by a node with no consumer for it. A lease’s expiry is now judged against the database’s own clock instead of the asking node’s, so clock skew between nodes can no longer shorten or extend a lease, and the builders’ clock(Clock) method no longer affects lease timing.
  • MongoProjectionStoreProvider’s save and saveAll now reject a @Projection state whose @Id doesn’t match the projection key with IllegalStateException, the same guard SpringMongoViewExtensions’s Mongo-backed repositories already applied. Before this, a mismatch silently wrote to the wrong document and the read model never accumulated.
  • Several CompetingConsumerStrategy beans with no @Primary now fail startup with AmbiguousCompetingConsumerStrategyException, which names the beans it found and what to do about them. Before this, Occurrent read the ambiguity as no strategy at all, so it wrote every checkpoint unconditionally and ran a @Saga’s timer poller on every instance. Mark the bean you want with @Primary, or leave only that one in the context. You probably do not have two, since the Mongo starter no longer contributes its default strategy once you register one of your own, whatever type it is. That also fixes a case that was quietly broken, where a custom strategy of any type other than SpringMongoLeaseCompetingConsumerStrategy never reached the subscription model at all and delivery kept running under the starter’s own lease.
  • A competing consumer that regains a lease it lost now resumes from the stored checkpoint instead of the position it had read before losing the lease. It no longer redelivers everything another node already handled while it was gone, and no longer writes the checkpoint backward. With EveryN(n) checkpointing the resume can still repeat up to n - 1 events, zero with the default everyEvent().
  • A MongoDB lease strategy’s scheduled refresh now gives up after 5 attempts against a down database instead of retrying forever. Refresh rounds run one at a time, so a round stuck on an unreachable database used to block every later round behind it, holding a lease past its deadline while a healthy rival waited to take over. Registering and unregistering a consumer still retry exactly as configured.

Full details are in the changelog, and the documentation has been updated with the new features.