Subscription annotations

Spring Boot · Added in 0.18.0

If using the Spring Boot Starter you have the option to start subscriptions using the @Subscription annotation (org.occurrent.annotation.Subscription). For example if you have a domain event like this:

Java

public sealed interface DomainEvent permits DomainEvent1, DomainEvent2, DomainEvent3 {
    UUID eventId();
    ZonedDateTime timestamp();
}

public record DomainEvent1(UUID eventId, ZonedDateTime timestamp, String someData1) implements DomainEvent {
}

public record DomainEvent2(UUID eventId, ZonedDateTime timestamp, String someData2) implements DomainEvent {
}

public record DomainEvent3(UUID eventId, ZonedDateTime timestamp, String someData3) implements DomainEvent {
}

Kotlin

sealed interface DomainEvent {
    val eventId: UUID
    val timestamp: ZonedDateTime
}

data class DomainEvent1( override val eventId: UUID, override val timestamp: ZonedDateTime, val someData1: String ) : DomainEvent
data class DomainEvent2( override val eventId: UUID, override val timestamp: ZonedDateTime, val someData2: String ) : DomainEvent
data class DomainEvent3( override val eventId: UUID, override val timestamp: ZonedDateTime, val someData3: String ) : DomainEvent

you can create a subscription to all events like this:

Java

@Component
public class Example {

    @Subscription(id = "printAllDomainEvents")
    void printAllDomainEvents(DomainEvent e) {
        System.out.println("Received domain event %s".formatted(e));
    }
}

Kotlin

@Component
class Example {

    @Subscription(id = "printAllDomainEvents")
    fun printAllDomainEvents(e: DomainEvent) {
        println("Received domain event $e")
    }
}

Note that subscriptions started by the Subscription annotation will make use of competing consumers so that if you run multiple instances of the application one of them will receive the event(s).

Since version 0.30.0, @Subscription is capability-neutral. On an event store that has both the stream and the DCB capability it delivers both stream-written and DCB-appended events, filtered by event type, with catch-up over the shared global position. Use @StreamSubscription when you want stream events only, and @DcbSubscription when you want DCB events only with tag-based filtering. @StreamSubscription is configured the same way as @Subscription described below, only scoped to stream events. Its startAt takes the nested StreamSubscription.StartPosition, where replaying from the start is BEGINNING_OF_TIME rather than BEGINNING.

All of these are asynchronous. For a handler that runs synchronously on the writer thread, before execute returns, and can commit atomically with the write, use @SynchronousSubscription instead. It is documented under Synchronous Subscriptions.

Subscription Start Position

When creating a subscription using the @Subscription annotation you can specify how it should behave when first starting (creating) the subscription as well as how it should be resumed when the application is restarted. Here’s an example:

Java

@Component
public class Example {

    @Subscription(id = "printAllDomainEvents", startAt = StartPosition.BEGINNING)
    void printAllDomainEvents(DomainEvent e) {
        System.out.println("Received domain event %s".formatted(e));
    }
}

Kotlin

@Component
class Example {

    @Subscription(id = "printAllDomainEvents", startAt = StartPosition.BEGINNING)
    fun printAllDomainEvents(e: DomainEvent) {
        println("Received domain event $e")
    }
}

This will first replay all historic events from the beginning of time and then continue subscribing to new events continuously. You can also start at a specific date by using startAtISO8601() or startAtTimeEpochMillis().

Note that the example above will start replaying historic events from the beginning of time when the subscription is started the first time. However, once the subscription is resumed, e.g. on application restart, it’ll continue from the last received event.

Here’s a description for each StartPosition:

StartPosition	Description
BEGINNING	Start this subscription from the first event in the event store.
NOW	Start this subscription from current time.
DEFAULT	Start this subscription using the default behavior of the subscription model. Typically, this means that it’ll start from NOW, unless the subscription has already been started before, in which case the subscription will be started from its last known checkpoint.

If you want a different behavior when the application is restarted, configure a different resumeBehavior() (@Subscription(id="mySubscription", resumeBehavior=..)):

Resume behavior:

ResumeBehavior	Description
DEFAULT	Use the default resume behavior of the underlying subscription model. For example, if the StartPosition is set to StartPosition.BEGINNING, and ResumeBehavior is set to ResumeBehavior.DEFAULT, then the subscription will start from the beginning of time the first time it’s run. Then, on application restart, it’ll continue from the last received event (the checkpoint for the subscription) on restart.
SAME_AS_START_AT	Always start at the same position as specified by the StartPosition. I.e., even if there’s a checkpoint stored for the subscription, it’ll be ignored on application restart and the subscription will resume from the specified StartPosition.

The combination of start and resume behavior can enable various use cases. For example:

StartPosition	ResumeBehavior	Use Case
BEGINNING	DEFAULT	Start a new subscription from “beginning of time”, but when the app is restarted, continue from the last processed event (i.e. don’t replay all historic events again).
BEGINNING	SAME_AS_START_AT	The subscription will always (even on restart) subscribe to all historic events, effectively making this an in-memory subscription. Note that this subscription will be called on each instance of the application even though competing consumer behavior is configured.
NOW	SAME_AS_START_AT	The subscription will always (even on restart) subscribe to events from “now”, ignoring the historic events. This will effectively make this an in-memory subscription. Note that this subscription will be called on each instance of the application even though competing consumer behavior is configured.
NOW	DEFAULT	The subscription start subscribing to events from “now” when created, but continue (resume) from the last received event on restart.
DEFAULT	DEFAULT	Same as above (this is the default behavior if start position and resume behavior is not specified).

Selective Events

If DomainEvent is a sealed interface/class (as in the examples above), then all events implementing this interface/class will be received when subscribing to this event. You can of course subscribe to an individual event, such as DomainEvent2. But if you want to receive only some of the events that implement the DomainEvent interface, you can use eventTypes(). For example, if you want to subscribe on both DomainEvent1 and DomainEvent3 but handle them as a DomainEvent:

Java

@Component
public class Example {

    @Subscription(id = "printAllDomainEvents", eventTypes = {DomainEvent1.class, DomainEvent3.class})
    void printSomeDomainEvents(DomainEvent e) {
        System.out.println("Received any of the specified domain events: %s".formatted(e));
    }
}

Kotlin

@Component
class Example {

    @Subscription(id = "printAllDomainEvents", eventTypes = [DomainEvent1::class, DomainEvent3::class])
    fun printSomeDomainEvents(e: DomainEvent) {
        println("Received any of the specified domain events: $e")
    }
}

The filter Occurrent derives from a sealed type names the declared type as well as the concrete types it permits. That only matters if you wrote a CloudEventTypeMapper that maps a whole hierarchy onto the type string of the type it was declared with, because such a subscription used to receive nothing at all. No mapper Occurrent ships stores an event under a sealed interface’s own type, so nothing changes for the default setup.

If you declare a type listed as refused in Deriving the Event Filter, Spring Boot startup fails. The message names the type and points you at eventTypes(). The same section has a table of when each of the other places refuses one.

Event Metadata

Sometimes it can be useful to get the metadata associated with the received event. For this reason, you can add a parameter of type org.occurrent.cloudevents.EventMetadata to a method annotated with @Subscription or @StreamSubscription. It contains all extension properties added to the CloudEvent, with typed accessors for the common ones: streamId, streamVersion, and position (the global sequence number, or null for a stream-written event on a store that does not record a position). Note that for an event delivered through the capability-neutral @Subscription or the DCB path, streamId is the internal generated partition id rather than a domain stream id, but it is always present. For example:

Java

@Component
public class Example {

    @Subscription(id = "printAllDomainEvents")
    void printAllDomainEvents(DomainEvent e, EventMetadata metadata) {
        String streamId = metadata.getStreamId();
        long streamVersion = metadata.getStreamVersion();
        Long position = metadata.getPosition();
        Object myCustomValue = metadata.get("MyCustomValue");
        ...
    }
}

Kotlin

@Component
class Example {

    @Subscription(id = "printAllDomainEvents")
    fun printAllDomainEvents(e: DomainEvent, metadata: EventMetadata) {
        val streamId: String = metadata.streamId
        val streamVersion: Long = metadata.streamVersion
        val position: Long? = metadata.position
        val myCustomValue: Any? = metadata["MyCustomValue"]
        // ...
    }
}

When you only need the stream id or stream version, annotate a handler parameter with @StreamId or @StreamVersion instead of taking the whole EventMetadata. @StreamId binds a String, @StreamVersion a long or Long, and they may appear in any order alongside the event and an optional EventMetadata parameter. This works on @Subscription, @StreamSubscription, and @SynchronousSubscription (the same internal-partition-id caveat above applies on the capability-neutral @Subscription). On @DcbSubscription these annotations are rejected at startup, since a DCB handler’s stream id and version are internal partition values rather than domain ones.

Java

@Component
public class Example {

    @Subscription(id = "printAllDomainEvents")
    void printAllDomainEvents(DomainEvent e, @StreamId String streamId, @StreamVersion long streamVersion) {
        ...
    }
}

Kotlin

@Component
class Example {

    @Subscription(id = "printAllDomainEvents")
    fun printAllDomainEvents(e: DomainEvent, @StreamId streamId: String, @StreamVersion streamVersion: Long) {
        // ...
    }
}

Subscription Startup Mode

You can configure whether the subscription should start before the application is ready to receive requests. For example, it might be very important that a certain subscription is started before the first web request comes in:

Java

@Component
public class Example {

    @Subscription(id = "printAllDomainEvents", startupMode = StartupMode.WAIT_UNTIL_STARTED)
    void printAllDomainEvents(DomainEvent e) {
        System.out.println("Received domain event %s".formatted(e));
    }
}

Kotlin

@Component
class Example {

    @Subscription(id = "printAllDomainEvents", startupMode = StartupMode.WAIT_UNTIL_STARTED)
    fun printAllDomainEvents(e: DomainEvent) {
        println("Received domain event $e")
    }
}

In other cases, such as when replaying a huge number of historic events it might be better to let the application start and let the processing of historic events happen in the background.

Java

@Component
public class Example {

    @Subscription(id = "printAllDomainEvents", startAt = StartPosition.BEGINNING, startupMode = StartupMode.BACKGROUND)
    void printAllDomainEvents(DomainEvent e) {
        System.out.println("Received domain event %s".formatted(e));
    }
}

Kotlin

@Component
class Example {

    @Subscription(id = "printAllDomainEvents", startAt = StartPosition.BEGINNING, startupMode = StartupMode.BACKGROUND)
    fun printAllDomainEvents(e: DomainEvent) {
        println("Received domain event $e")
    }
}

Here’s a summary of the different startup modes:

StartupMode	Description
DEFAULT	Determine the startup mode based on the properties of the subscription (such as startAt() and resumeBehavior()). It’ll use BACKGROUND if the subscription needs to replay historic events before subscribing to new ones (e.g. if startAt() is StartPosition.BEGINNING_OF_TIME), otherwise WAIT_UNTIL_STARTED will be used.
WAIT_UNTIL_STARTED	The subscription will wait until it’s started up fully before Spring continues starting the rest of the application. Most of the time this is recommended because otherwise there could be a small chance that a request is received by your application before the subscription has bootstrapped completely. This can lead to the subscription missing this event. This is only true if the subscription is brand new. As soon as the subscription has received an event that is stored in a org.occurrent.subscription.api.blocking.CheckpointStorage, it’ll never miss an event during startup.
BACKGROUND	The subscription will NOT wait until it’s started up fully before Spring continues starting the rest of the application; instead, it will be started in the background. Typically, this is useful if you instruct the subscription to start at an earlier date (such as the beginning of time), and you have a lot of events to read before the subscription has caught up. In this case, you may wish to start the Spring application before the subscription has fully started (i.e., before all historic events have been replayed) because waiting for all events to replay takes too long. The subscription will then replay all historic events in the background before switching to continuous mode.

These modes decide what happens while the application starts. The subscription handlers of a bean built after startup never wait for their replay, whatever startupMode says, see When Annotations Register.

Spring Advice on Handler Methods

Occurrent calls a handler method through the bean’s Spring proxy, so @Transactional, or any other aspect on the method or its class, applies to every event the method receives:

@Component
public class OrderStatusUpdater {

    @Transactional
    @Subscription(id = "orderStatus")
    public void onOrderPlaced(OrderPlaced orderPlaced) {
        // runs inside a transaction
    }
}

This applies to @Subscription, @StreamSubscription, @DcbSubscription and @SynchronousSubscription alike.

If the proxy can’t reach a handler method, Spring Boot startup fails with SubscriptionHandlerNotInvocableException instead of running the method without its advice:

Handler method	Why the proxy can’t reach it	Fix
Declared only on the class, on a bean behind a JDK interface proxy	The proxy implements only the interfaces	Declare the method on an interface, or set spring.aop.proxy-target-class=true
private, on a bean behind a CGLIB proxy	A CGLIB proxy can’t override a private method	Make the method non-private
final, on a bean behind a CGLIB proxy	A CGLIB proxy can’t override a final method	Remove final
static	Calling it never goes through any proxy	Make it an instance method

A private or final handler on a bean that isn’t proxied at all still runs, since there’s no advice to lose.

When Annotations Register

Every @Subscription, @StreamSubscription, @DcbSubscription, @SynchronousSubscription, @Projection, @Snapshot and @Saga registers after all singleton beans have been created and before any ApplicationRunner or CommandLineRunner runs. A subscription that only receives new events therefore sees an event written by an ApplicationRunner, but misses one written while the beans are still being created, from a @PostConstruct method for example.

A @Lazy bean whose declared type shows one of these annotations is built at that point too, so it registers with the others.

A bean whose declared type doesn’t show the annotation registers when something first builds it. That happens when a @Lazy @Bean method returns an interface and the annotated method is only on the class that implements it:

@Configuration
public class NotifierConfig {

    @Bean
    @Lazy
    public OrderNotifier orderNotifier() { // OrderNotifier is an interface
        return new EmailOrderNotifier(); // @Subscription is declared on EmailOrderNotifier
    }
}

A FactoryBean registers the same way when nothing has asked for its object yet. A bean that nothing ever builds never registers its handlers.

A @Subscription, @StreamSubscription or @DcbSubscription registered this way doesn’t wait for its replay, whatever startupMode says, since the application is already running. On the reactor stack, the same goes for a @Projection or @Snapshot.

A Bean Built on a Reactor Non-Blocking Thread

On the reactor stack the bean can be built on a Reactor non-blocking thread, for example when a WebFlux handler asks for it:

Mono.fromCallable(() -> applicationContext.getBean(OrderNotifier.class))
        .subscribeOn(Schedulers.parallel());

Reactor doesn’t allow blocking on such a thread. Subscribing a handler that starts at the beginning or at an explicit position can block while the starter reads the position stored for it. On any other thread, every handler subscribes on the thread that builds the bean. On a non-blocking thread, it depends on where the handler starts:

Start position	What happens
The beginning, or an explicit position	Subscribes on Schedulers.boundedElastic().
NOW or DEFAULT	Subscribes on the thread that builds the bean, which works on the model the starter builds.

A handler that starts at NOW or DEFAULT subscribes on the calling thread because it starts from wherever the event feed is when it subscribes. Subscribing it later could skip what the caller writes in between. A handler that starts at the beginning or at an explicit position receives the same events no matter when it subscribes.

When a subscription model other than the one the starter builds blocks while subscribing a NOW or DEFAULT handler, the bean fails to build instead, and the exception message says what to do. To fix it, build the bean on a thread that may block, with Schedulers.boundedElastic() instead of Schedulers.parallel() above, or start the handler at the beginning or at an explicit position.

When the subscribe runs on Schedulers.boundedElastic(), there’s no caller to throw an exception to, so a failure there is logged at ERROR. After a failure that might be temporary, an unreachable storage for example, Occurrent tries the subscribe again until an attempt succeeds or the application context closes. The delay between attempts doubles from 100 ms up to 30 seconds, and the handler receives no events until an attempt succeeds.

Some failures happen the same way on every attempt. Occurrent logs these once and gives up on the subscription. They include:

• a duplicate subscription id, or a filter or start position the model refuses
• a bean the context doesn’t have, has with another type, or has only as an abstract definition
• a subscription model that was shut down
• an Error

If a bean the subscribe needs, CheckpointStorage for example, failed to build, Occurrent looks at the root cause of that build failure to decide whether to try again.

The handlers on one bean subscribe one after the other. So when Occurrent gives up on one handler, it also gives up on the handlers after it, and the log names them. For a singleton bean, fix the cause and restart the application, since Spring builds a singleton only once. For a bean that isn’t a singleton, the next instance Spring builds tries to register its handlers again.