Persistent Streams

A Subscribing Processor can use a persistent stream as its event source. By using a persistent stream, a Subscribing Processor can process events in parallel and replay events.

When a processor uses a persistent stream, it receives events from Axon Server. After processing a batch of events, it sends an acknowledgment back to Axon Server.

The persistent stream can be split into segments to allow for parallel processing within a single client or across multiple instances of the client. The number of segments can be changed dynamically. Axon Server distributes the segments across the subscribers to ensure that all segments are connected.

Events are assigned to a specific segment based on the sequencing policy for the persistent stream. Persistent streams support all the standard sequencing policies that can also be used for streaming processors.

If no sequencing policy is specified for a persistent stream, the configuration defaults to SequentialPolicy. This policy instructs the persistent stream to process all events sequentially. Be sure to explicitly check and configure your sequencing policy to match your specific sequencing requirements. If you are using an aggregate-based approach with a non-DCB context in Axon Server, consider using the SequentialPerAggregatePolicy instead of the SequentialPolicy.

Clients can provide a filter in the persistent stream definition. This reduces the number of events that the client receives from Axon Server. The filter expression uses the same syntax as the ad-hoc query option in Axon Server.

Persistent streams do not require a token store in the client like streaming processors do. The state of the stream is maintained in Axon Server.

Configuration

For a specific Event Processor to use a persistent stream as its source, the event source must be a PersistentStreamEventSource. You can configure this in three ways: declaratively through the Configuration API, through Spring Boot properties or autodetected via Spring Boot EventProcessorDefinition beans.

See the examples below, or consult the general processor configuration section for further options.

Each persistent stream must be identified by a unique name within your Axon Server environment. The PersistentStreamEventSource requires a PersistentStreamProperties to define the initial stream creation settings:

streamName: the unique identifier of the persistent stream in Axon Server; reusing an existing name joins the same server-side stream.
segments: the initial number of segments for parallel processing.
sequencingPolicyName: the name of the sequencing policy; use the constants on PersistentStreamSequencingPolicy (for example, SEQUENTIAL_PER_AGGREGATE_POLICY).
sequencingPolicyParameters: list of parameters for the sequencing policy; required for PropertySequencingPolicy and MetadataSequencingPolicy.
initialPosition: the position to start reading from, which is either a global event sequence number, "HEAD" (start from the latest event), or "TAIL" (start from the beginning of the event store).
filter: an optional server-side filter expression in Axon Server Query Language; use null to receive all events.

For the MetadataSequencingPolicy, the sequencingPolicyParameters must contain the name of one or more event metadata fields. Events with the same value for these fields are routed to the same segment.

The PropertySequencingPolicy requires 4 values in the sequencingPolicyParameters list:

The serialization format of the events: JSON or XML (currently only JSON or XML supported).
The payload type to apply the policy on.
An expression to extract the sequencing value from the event payload (JsonPath for JSON, XPath for XML).
A fallback policy name to use when the event payload type does not match; this may be another PropertySequencingPolicy to chain additional payload-type rules.

Declarative configuration

When using Axon’s Configuration API directly, retrieve the infrastructure components via the Configuration parameter and pass them to the PersistentStreamEventSource constructor:

import io.axoniq.axonserver.connector.event.PersistentStreamProperties;
import io.axoniq.framework.axonserver.connector.event.PersistentStreamEventSource;
import io.axoniq.framework.axonserver.connector.event.PersistentStreamEventSourceFactory;
import io.axoniq.framework.axonserver.connector.event.PersistentStreamScheduledExecutorBuilder;
import io.axoniq.framework.axonserver.connector.event.PersistentStreamSequencingPolicy;
import org.axonframework.common.configuration.Configuration;
import org.axonframework.messaging.core.configuration.MessagingConfigurer;
import org.axonframework.messaging.eventhandling.configuration.EventHandlingComponentsConfigurer;
import org.axonframework.messaging.eventhandling.processing.subscribing.SubscribingEventProcessorsConfigurer;

import java.util.Collections;

public class AxonConfig {

    public void configureEventProcessing(MessagingConfigurer messaging) {
        messaging.eventProcessing(
                eventProcessing -> eventProcessing.subscribing(
                    this::configureSubscribingProcessor
        ));
    }

    private SubscribingEventProcessorsConfigurer configureSubscribingProcessor(
            SubscribingEventProcessorsConfigurer subscribingConfigurer
    ) {
        return subscribingConfigurer.processor(
                "example-processor",
                config -> config.eventHandlingComponents(this::configureHandlingComponent)
                        .customized((c, subscribingConfig) -> subscribingConfig.eventSource(
                                buildPersistentStreamSource(c)
                        ))
        );
    }

    private PersistentStreamEventSource buildPersistentStreamSource(Configuration c) {
        String streamName = "example-persistent-stream";
        int segmentCount = 4;
        int batchSize = 1024;

        PersistentStreamProperties properties = new PersistentStreamProperties(
                streamName,
                segmentCount,
                PersistentStreamSequencingPolicy.SEQUENTIAL_POLICY,
                Collections.emptyList(),
                "HEAD",
                null  // no server-side filter
        );

        return PersistentStreamEventSourceFactory.defaultFactory()
                .build(streamName,
                        properties,
                        PersistentStreamScheduledExecutorBuilder.defaultFactory().build(segmentCount, streamName),
                        batchSize,
                        c);
    }

    private EventHandlingComponentsConfigurer.AdditionalComponentPhase configureHandlingComponent(
            EventHandlingComponentsConfigurer.RequiredComponentPhase componentConfigurer
    ) {
        return componentConfigurer.autodetected(c -> new AnnotatedEventHandlingClass());
    }
}

Properties-based configuration - Spring Boot properties file

The simplest way to configure a persistent stream in Spring Boot is through the properties file. Each entry under axon.axonserver.persistent-streams registers a PersistentStreamEventSource Spring bean named after the map key. The source value of the processor must match that key exactly. See the io.axoniq.framework.axonserver.connector.api.AxonServerConfiguration.PersistentStreamSettings Java class for all available persistent stream configuration properties.

axon.eventhandling.processors.example-processor.mode=subscribing
axon.eventhandling.processors.example-processor.source=example-persistent-stream

axon.axonserver.persistent-streams.example-persistent-stream.name=Example Persistent Stream
axon.axonserver.persistent-streams.example-persistent-stream.batch-size=100
axon.axonserver.persistent-streams.example-persistent-stream.initial-segment-count=4

Multiple persistent streams can be declared side-by-side under different keys, and each processor declares its own source to resolve unambiguously:

axon.eventhandling.processors.payments-processor.mode=subscribing
axon.eventhandling.processors.payments-processor.source=payments-stream

axon.eventhandling.processors.orders-processor.mode=subscribing
axon.eventhandling.processors.orders-processor.source=orders-stream

axon.axonserver.persistent-streams.payments-stream.name=Payments Stream
axon.axonserver.persistent-streams.payments-stream.initial-segment-count=4

axon.axonserver.persistent-streams.orders-stream.name=Orders Stream
axon.axonserver.persistent-streams.orders-stream.initial-segment-count=2

If you want to default all processors to use persistent streams, you could use the following shorthand configuration:

axon.axonserver.auto-persistent-streams-enabled=true

This automatically configures a persistent stream named [processor]-stream for each processor that is not explicitly defined as streaming processor in your configuration. To customize the auto-configured persistent streams further you can use the following configuration properties. Same as for individual persistent stream configurations, see the io.axoniq.framework.axonserver.connector.api.AxonServerConfiguration.PersistentStreamSettings Java class for all available persistent stream configuration properties.

axon.axonserver.auto-persistent-streams-settings.batch-size=5
axon.axonserver.auto-persistent-streams-settings.thread-count=10
axon.axonserver.auto-persistent-streams-settings.initial-segment-count=10

Autodetected configuration - `EventProcessorDefinition` based configuration

When using Spring Boot, you can wire a persistent stream to a processor through an EventProcessorDefinition bean. The definition selects the event handlers for the processor and supplies the PersistentStreamEventSource as its event source in the customized(…) step.

import io.axoniq.axonserver.connector.event.PersistentStreamProperties;
import io.axoniq.framework.axonserver.connector.event.PersistentStreamEventSourceFactory;
import io.axoniq.framework.axonserver.connector.event.PersistentStreamScheduledExecutorBuilder;
import io.axoniq.framework.axonserver.connector.event.PersistentStreamSequencingPolicy;
import org.axonframework.common.configuration.AxonConfiguration;
import org.axonframework.extension.spring.config.EventHandlerSelector;
import org.axonframework.extension.spring.config.EventProcessorDefinition;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.Collections;
import java.util.concurrent.ScheduledExecutorService;

@Configuration
class EventProcessorConfig {

    private static final String STREAM_NAME = "example-persistent-stream";
    private static final int SEGMENT_COUNT = 4;

    @Bean(destroyMethod = "shutdown") (1)
    public ScheduledExecutorService exampleStreamExecutor(
            PersistentStreamScheduledExecutorBuilder executorBuilder) {
        return executorBuilder.build(SEGMENT_COUNT, STREAM_NAME);
    }

    @Bean
    public EventProcessorDefinition exampleProcessorDefinition(
            AxonConfiguration configuration,
            PersistentStreamEventSourceFactory eventSourceFactory,
            ScheduledExecutorService exampleStreamExecutor) {
        int batchSize = 1024;

        PersistentStreamProperties properties = new PersistentStreamProperties(
                STREAM_NAME,
                SEGMENT_COUNT,
                PersistentStreamSequencingPolicy.SEQUENTIAL_PER_AGGREGATE_POLICY,
                Collections.emptyList(),
                "HEAD",
                null  // no server-side filter
        );

        return EventProcessorDefinition.subscribing("example-processor")
                .assigningHandlers(EventHandlerSelector.matchesNamespaceOnType("orders"))
                // Build the event source inside the lambda: it runs once the Axon configuration
                // is fully initialized. Querying the AxonConfiguration during bean construction
                // fails the application start-up.
                .customized(config -> config.eventSource(
                        eventSourceFactory.build(STREAM_NAME,
                                                 properties,
                                                 exampleStreamExecutor,
                                                 batchSize,
                                                 configuration))); (2)
    }
}

1	Registering the executor as a Spring bean with a `shutdown` destroy method ensures its threads are released when the application context closes.
2	Building the event source inside the `customized(…)` lambda to ensure the `AxonConfiguration` is fully initialized and allows retrieving the required dependencies.

The creation of the StreamableEventSource deliberately happens inside the lambda passed to customized(…). The lambda only runs when the event processor is created at application start-up, after the Axon configuration has been fully initialized.

Building the event source outside the lambda, for example in a separate @Bean method, looks up components while the application context is still starting and fails the boot process.

Replaying events

Persistent stream backed event processors allow replaying events in a similar way as streaming event processors do. However, due to the nature of subscribing processors and the server-side implementation of persistent streams there are a few limitations compared to streaming processors:

the ReplayToken that might be accessed from the processing context or as annotated handler parameter always reports the next token position as the token at reset instead to the real position at which the reset was triggered (so a replay startet a position 10 will have a replay token for an event at position 5 report position 6 as token at reset, the event at position 6 will report position 7 as token at reset)
@ReplayContext annotated handler parameters are not supported (they will be always null)
@ResetHandler annotated methods (used to run pre-reset logic) are not supported (they will simply be ignored)