Occurrent 0.34.0 is released. It can forward stored events to RabbitMQ or Kafka and consume them back into a push subscription, and it removes the flow saga’s deprecated join. Below are the highlights, a few smaller additions, and the changes to read before you upgrade.

Highlights

Forward events to RabbitMQ or Kafka

The new occurrent-broker-api-blocking module forwards stored events to a broker at least once. CloudEventForwarder runs a durable subscription and hands each stored event to a CloudEventSink (the class that publishes an event to the broker). The forwarder saves its checkpoint, the record of how far it has got, only after the sink has returned. If the sink throws, the same event is published again on the next run. DomainEventForwarder does the same for an application whose converter already produces domain events.

occurrent-broker-rabbitmq-blocking and occurrent-broker-kafka-blocking each ship a sink and a source. The sink publishes events to the broker. A RabbitMQ publish returns only after the message is both confirmed and routed, and a Kafka publish only after the broker has acknowledged it with acks=all.

A source is a class that consumes messages from the broker and hands each one to your subscriptions. RabbitMqCloudEventSource and KafkaCloudEventSource hand it to a PushSubscriptionModel as a CloudEvent, and RabbitMqDomainEventSource and KafkaDomainEventSource hand it to a DomainEventFeed, which decodes it into your domain event.

A source acknowledges a message once your subscription has handled it or its filter has skipped it, and keeps it on the broker while the subscription isn’t ready for it yet. When the handler throws, your DeliveryFailurePolicy decides whether the message is redelivered or parked. If the subscription’s catch-up replay has failed for good, the source stops consuming. Consuming from a broker shows how to build a source.

With Spring Boot, @EnableOccurrentRabbitMqBroker or @EnableOccurrentKafkaBroker builds the sink and a source factory from the occurrent.broker.rabbitmq.* or occurrent.broker.kafka.* properties. The broker modules only exist for the blocking stack. The documentation shows the wiring for each broker, and example/broker/rabbitmq in the Occurrent repository runs the whole loop against RabbitMQ and MongoDB.

Sealed event types match all their subtypes

A sealed event supertype now matches every concrete type it permits. This applies when you declare it in a projection, a subscription built through the subscription DSL, a DomainEventQueries query, a @Snapshot view, and DcbCriteriaBuilder’s type or types. A saga and an annotation-based subscription already worked this way.

It also applies to ExecuteFilter.type and includeTypes. An ExecuteFilter is what you pass to ApplicationService.execute(..) when the application service should load only some of the events in a stream. When ExecuteFilter.excludeTypes can’t find every concrete subtype of a declared type, it excludes every one it can find, because excluding a supertype has to exclude everything under it. The documentation says how a declared type becomes a filter.

Acknowledging messages in your own broker listener

A broker listener is the code that receives messages from RabbitMQ, Kafka or another broker, and tells the broker when it’s done with each one so the broker doesn’t send it again. That last step is called acknowledging. The sources above are broker listeners that Occurrent ships.

If you write your own, turn each message into a CloudEvent and pass it to PushSubscriptionModel.acceptOrDefer(CloudEvent). It returns a RoutingOutcome, which tells you whether to acknowledge the message. The reactive PushSubscriptionModel has the same method, returning a Mono<RoutingOutcome>.

DELIVERED means a running subscription accepted the event, and FILTERED means its filter declined it. The other four, UNAVAILABLE, NOT_DELIVERABLE, DEFERRED and REFUSED, say why nothing consumed the event, for example because nothing was subscribed or a replay was still running.

Switch on outcome.disposition(), which is ACKNOWLEDGE, HOLD, FAIL or STOP. HOLD means don’t acknowledge the message yet, and hand it back to the broker so it’s delivered again. If a yes or no is enough, outcome.mayAcknowledge() is true only for DELIVERED and FILTERED.

When the handler or the subscription’s filter fails, acceptOrDefer(..) throws, or its Mono errors. Don’t acknowledge the message then either, and hand it back the same way or move it to a failed-message queue.

PushSubscriptionModel also takes an optional PushObserver, on both stacks, which gets the outcome of every event passed to accept(..) or acceptOrDefer(..). A misconfigured queue binding or a typo in a type mapping then shows up instead of going unnoticed. Don’t decide whether to acknowledge from what the observer gets, because it gets DELIVERED even when the handler threw. The documentation says what each outcome means.

A saga instance that keeps failing can be quarantined

Up to 0.33.0, when a saga’s evolve, react or command dispatcher failed on an event, the instance failed again every time the event was redelivered, forever. On the MongoDB subscription models, the saga’s other instances were stuck behind that event the whole time.

In 0.34.0 you can set a time limit with SagaRunnerConfig.withQuarantineAfter(Duration), or occurrent.saga.quarantine-after with Spring Boot. Once an instance has kept failing for that long, Occurrent can quarantine it, which sets its status to SagaStatus.QUARANTINED. A quarantined instance skips every event sent to it, including the one it failed on, and none of its timers fire. The saga’s other instances get their events as usual.

There’s no time limit by default, because nothing in 0.34.0 brings a quarantined instance back. Deleting it with SagaStateStore.delete(sagaId) is the only way out.

Quarantine also needs a subscription model that keeps every event it delivers. The MongoDB subscription models do, including when a catch-up or competing-consumer model wraps one, and on a model that doesn’t, a saga is never quarantined and logs a warning when it starts if you set the limit.

The documentation covers how to find a quarantined instance, and the runbook walks through deciding what to do with one.

An asynchronous projection can wait for its own write

Every write and DCB append now returns an append id, as an Optional<AppendId> on its result, and a projection can record which ones it has applied. Wait for a specific append with AppliedAppendStore.waitUntilApplied(...) instead of running the whole projection synchronously just to see your own write.

With the Spring Boot starter, set recordAppliedAppends = true on @Projection and the starter sets up the store and the recording for you. See the documentation for the wiring without Spring Boot, and what the wait does and doesn’t guarantee.

MongoDB subscriptions save their position even when no events match

Up to 0.33.0, a MongoDB subscription’s position moved only when an event matched its filter. After a long period with no matching events, a pause, a lease handover or a restart opened the change stream at a position the oplog no longer had. The subscription was treated as if it had lost history, although it had missed nothing.

Now the MongoDB subscription models report how far a subscription has read even when no event matched, and DurableSubscriptionModel and ReactorDurableSubscriptionModel save that position at most once a minute. On the reactor stack this needs a subscription with an id and a MongoDB driver the model can read the resume token from, and the model logs a warning when it’s created if it can’t.

Change the interval with saveQuietPositionEvery(Duration) on the durable model’s config, and keep it well below the oplog window. neverSaveQuietPosition() turns the save off. With a persist predicate that declines some events, EveryN with n above 1 for example, nothing is saved after a declined event until the predicate stores one.

Smaller additions

  • Projection.builder(), Projection.singletonBuilder(), Saga.builder(), SnapshotView.builder() and View.create gain a no-argument overload for state that starts empty, so you no longer pass null. In Kotlin the state is then S?, so a handler that uses it without a null check fails to compile instead of throwing on the first event.
  • @Transactional, or any other Spring aspect, on a @Subscription, @StreamSubscription, @DcbSubscription or @SynchronousSubscription handler method now applies to every delivery, including one delivered while Spring Boot is still starting.
  • SpringMongoSagaStateStore retries a transient MongoDB error, with backoff from 100 ms up to 2 seconds and at most 10 attempts. A MongoDB error that goes away before the attempts run out no longer reaches the saga as a failed transition, so it doesn’t count toward the quarantine budget.
  • A @Projection, @Saga or @Snapshot on a bean Spring Boot builds only after startup, a @Lazy bean for example, now registers when the bean is built. In 0.33.0 it never registered.
  • ReactorMongoEventStore.updateEvent no longer logs the whole stored document, data payload included, at INFO.
  • DomainEventFeed gains acceptOrDeferCloudEvent(CloudEvent) on both stacks, for a broker listener that has a CloudEvent rather than a decoded domain event. It decodes only an event that matches the filter register(..) was given, and returns its RoutingOutcome. acceptAndWaitCloudEvent(CloudEvent) matches the same way and waits for the catch-up instead, as acceptAndWait(..) does.
  • RetryStrategy.execute(..) takes a shutdown predicate, retryStrategy.execute(runnable, __ -> running) for example, so a component that is closing down can stop a retry loop that is already running, even while it waits between two attempts. A RetryStrategy you implement yourself ignores the predicate.

Breaking changes

Run the org.occurrent.UpgradeToOccurrent_0_34 recipe first. It makes the mechanical changes, including some of the ones below.

API changes

The flow saga’s join is removed, together with Kotlin’s expect<T> and Expectation. Use on(StepCondition, ...) with allOf(...) instead. The recipe rewrites a join built from a literal list of literal Expectation.of(...) calls, and the compiler finds the rest, including every Kotlin call site (section 1).

WriteResult and DcbAppendResult gain a fourth component, Optional<AppendId> appendId(). A record pattern naming the original three stops compiling, and the recipe adds the fourth binding. An assertion comparing a whole record still compiles but fails, because every write that persists something gets a fresh append id, so assert on the components you mean (section 6).

SagaStatus gains QUARANTINED, even if you never turn quarantine on. SagaInstance also gains failure(), SagaEnvelope gains started and failure, and SagaRunnerConfig gains quarantineAfter. That breaks an exhaustive switch or when over SagaStatus, your own SagaInstance class, and a record pattern over SagaEnvelope or SagaRunnerConfig, and the recipe doesn’t fix any of them. findByStatus(ACTIVE, ...) doesn’t return a quarantined instance, so a stuck-instance check has to ask for QUARANTINED too (section 8).

SpringMongoSubscription loses its protected constructor. SpringMongoSubscriptionModel now reads change streams with the same cursor loop as NativeMongoSubscriptionModel instead of Spring Data’s MessageListenerContainer, so a subclass or code that creates a SpringMongoSubscription stops compiling (section 19).

A reactor CheckpointAwareSubscriptionModel of your own must implement globalCheckpointAsOfNow(), which has no default. Section 21 says how to answer it.

The reactor cancelSubscription(..) returns a Mono<Void>, on CancellableSubscriptions and DcbSubscriptionModel, and so does DcbSubscriptions.cancel(..). It completes once the state stored for that id is deleted, so wait for it before you subscribe the same id again. The recipe changes a Java implementation that returns void, and section 23 lists what you still change by hand.

Behavior changes

  • A flow saga’s stepWindow counts only events of the types its steps declare, plus repeats of the type that starts the flow. See section 2.
  • A declared event type whose concrete subtypes can’t be found is refused everywhere a filter is derived from it, and so is a concrete class that is neither final nor sealed. See section 3.
  • Four MongoDB-only Spring Boot keys move under mongodb, occurrent.event-store.collection to occurrent.event-store.mongodb.collection for example. The old keys still work but are deprecated. See section 4.
  • A @Projection, @Saga or @Snapshot bean’s class-level advice no longer runs once at startup, and a reactor factory that returns null fails startup with IllegalStateException. See section 5.
  • DurableSubscriptionModel refuses a first subscription when the subscription model it wraps can’t give it a start position, which is the case on a shared MongoDB Atlas cluster. See section 7.
  • A reactor catch-up subscription can deliver an event written during its replay twice. See section 9.
  • Events that updateEvent rewrote on 0.33.0 or earlier have a string position and no dcbTags, so reads and conditional appends skip them without an error, and they need a one-off repair. See section 10.
  • A subscription handler that Spring’s proxy can’t invoke fails startup, and annotation-based subscription handlers register only after every singleton has been created. See section 11.
  • A projection feed’s accept(..) is deprecated on both stacks and forwards to the new acceptAndWait(..), and the UpgradeToOccurrent_0_34 recipe renames each call. On the blocking stack acceptAndWait(..) waits until the event is applied, and throws when it isn’t. A listener that shouldn’t wait calls the new acceptOrDefer(..), which returns DEFERRED until the feed is live. See section 12.
  • PushSubscriptionModel.accept(..) is supported only from an InMemoryEventStore listener. See section 13.
  • A projection feed’s goLive() waits for a running catch-up of the same projection. See section 14.
  • Feeding a reactor catch-up from its own handler completes once the event is queued instead of waiting forever. See section 15.
  • A NativeMongoSubscriptionModel subscription made while the model is stopped opens no change stream until start() or a resume. See section 16.
  • A blocking catch-up from StartAtTime.offsetDateTime(time) also replays the events stored at time. See section 17.
  • CompetingConsumerSubscriptionModel changes how it starts, reports and shuts down. See sections 18, 20, 22 and 24.
  • SpringMongoSubscriptionModel no longer skips an event whose action keeps failing, and a durable subscription whose filter you widen resumes from its quiet position. See section 19.
  • A ReactorMongoSubscriptionModel subscription started at the present starts from the subscribe(..) call. See section 21.
  • A durable reactor subscribe(..) from StartAt.subscriptionModelDefault() returns without waiting for storage, and waitUntilStarted() reports what it refuses. See section 25.
  • Before its replay, a position catch-up reads the change-stream position that live delivery picks up from, and stores it in its checkpoint. A storage that keeps strings stores a format 0.33.0 can’t read. See section 26.
  • When MongoDB’s oplog drops the position a catch-up goes live from just before its change stream opens there, the MongoDB subscription model asks the catch-up first, whatever occurrent.subscription.mongodb.restart-on-change-stream-history-lost says. The catch-up replays again from where its replay started and delivers what it had already replayed a second time, instead of the model going live from the present and skipping the events in between. See the documentation and section 26.
  • A DCB-only MongoDB event store no longer creates the dcbTags index. See section 27.
  • A blocking time-based catch-up now reads the change-stream position that live delivery picks up from before its replay, and stores it in a CatchupTimeCheckpoint. An event with an earlier time that commits during the replay, or while the catch-up is stopped mid-replay, is no longer lost. More events can arrive twice, the oplog has to reach back over the time replay, and a storage that keeps strings stores a format 0.33.0 can’t read. See the documentation and section 28.
  • On the blocking stack, a node that gets the lease of a subscription whose catch-up another node hadn’t finished replays from the position that catch-up stored. In 0.33.0 it went live from the present and skipped the events in between. Events the other node already delivered after that position now arrive again, and the resume throws IllegalStateException over a wrapped model that can’t resume from a given position. See the documentation and section 29.

Full details are in the changelog, and the upgrade guide walks through every change with the by-hand translation where the recipe cannot reach. Upgrading to 0.34.0 in the documentation has a shorter version.