Streaming Event Processors

The StreamingEventProcessor, or Streaming Processor for short, is a type of Event Processor. As any Event Processor, it serves as the technical aspect to handle events by invoking the event handlers written in an Axon application.
The Streaming Processor defines itself by receiving the events from a StreamableMessageSource. The StreamableMessageSource is an infrastructure component through which we can open a stream of events. The source can also specify positions on the event stream, so-called Tracking Tokens, used as start positions when opening an event stream. An example of a StreamableMessageSource is the EventStore, like for example Axon Server or an RDBMS.
Furthermore, Streaming Processors use separate threads to process the events retrieved from the StreamableMessageSource. Using separate threads decouples the StreamingEventProcessor from other operations (e.g., event publication or command handling), allowing for cleaner separation within any application.
Using separate threads allows for parallelization of the event load, either within a single JVM or between several.
When starting a Streaming Processor, it will open an event stream through the configured StreamableMessageSource. The first time a stream has started, it, by default, will begin at the tail (the oldest/the very first token) of the stream. It keeps track of the event processing progress while traversing the stream. It does so by storing the Tracking Tokens, or tokens for short, accompanying the events. This solution works towards tracking the progress since the tokens specify the event's position on the stream.
Head or Tail?
The oldest (very first) token is located at the tail of the stream, and the latest (newest) token is positioned at the head of the stream.
Maintaining the progress through tokens makes a Streaming Processor
1.able to deal with stopping and starting the processor,
2.more resilient against unintended shutdowns, and
3.the token provides a means to replay events by adjusting the position of tokens.
All combined, the Streaming Processor allows for decoupling, parallelization, resiliency, and replay-ability. It is these features that make the Streaming Processor the logical choice for the majority of applications. Due to this, the "Tracking Event Processor," a type of Streaming Processor, is the default Event Processor.
Default Event Processor
Which EventProcessor type becomes the default processor depends on the event message source available in your application. In the majority of use cases, an Event Store is present. As the Event Store is a type of StreamableMessageSource, the default will switch to the Tracking Event Processor.
If the application only has an Event Bus configured, the framework will lack a StreamableMessageSource. It will fall back to the Subscribing Event Processor as the default in these scenarios. This implementation will use the configured EventBus as its SubscribableMessageSource.
There are two implementations of Streaming Processor available in Axon Framework:
1.the Tracking Event Processor (TEP for short), and
2.the Pooled Streaming Event Processor (PSEP for short).
Both implementations support the same set of operations. Operations like replaying events through a reset, parallelism and tracking the progress with tokens. They diverge on their threading approach and work separation, as discussed in more detail in this section.
Configuring
The Streaming Processors have several additional components that you can configure, next to the base options. For other streaming processor features that are configurable, we refer to their respective sections for more details. This chapter will cover how to configure a Tracking or Pooled Streaming Processor respectively.
Configuring a Tracking Processor
Firstly, to specify that new event processors should default to a TrackingEventProcessor, you can invoke the usingTrackingEventProcessors method:
Axon Configuration API
Spring Boot AutoConfiguration
public class AxonConfig { 
    // omitting other configuration methods...
    public void configureProcessorDefault(EventProcessingConfigurer processingConfigurer) { 
        processingConfigurer.usingTrackingEventProcessors();  
    }
}
@Configuration
public class AxonConfig {
    // omitting other configuration methods...
    @Bean
    public ConfigurerModule processorDefaultConfigurerModule() {
        return configurer -> configurer.eventProcessing(EventProcessingConfigurer::usingTrackingEventProcessors);
    }
}
For a specific Event Processor to be a Tracking instance, registerTrackingEventProcessor is used:
Axon Configuration API
Spring Boot AutoConfiguration - Java
Spring Boot AutoConfiguration - Properties file
public class AxonConfig {
    // omitting other configuration methods...
    public void configureTrackingProcessors(EventProcessingConfigurer processingConfigurer) {
        // This configuration object allows for fine-grained control over the Tracking Processor
        TrackingEventProcessorConfiguration tepConfig =
              TrackingEventProcessorConfiguration.forSingleThreadedProcessing();
        
        // To configure a processor to be tracking ...
        processingConfigurer.registerTrackingEventProcessor("my-processor")
                            // ... to define a specific StreamableMessageSource ... 
                            .registerTrackingEventProcessor(
                                    "my-processor", conf -> /* create/return StreamableMessageSource */
                            )
                            // ... to provide additional configuration ...
                            .registerTrackingEventProcessor(
                                    "my-processor", conf -> /* create/return StreamableMessageSource */,
                                    conf -> tepConfig
                            );
    }
}
@Configuration
public class AxonConfig {
    // omitting other configuration methods...
    @Bean
    public ConfigurerModule trackingProcessorConfigurerModule() {
        // This configuration object allows for fine-grained control over the Tracking Processor
        TrackingEventProcessorConfiguration tepConfig =
              TrackingEventProcessorConfiguration.forSingleThreadedProcessing();
​
        return configurer -> configurer.eventProcessing(
              // To configure a processor to be tracking ...
              processingConfigurer -> processingConfigurer.registerTrackingEventProcessor("my-processor")
                                                          // ... to define a specific StreamableMessageSource ...
                                                          .registerTrackingEventProcessor(
                                                                  "my-processor",
                                                                  conf -> /* create/return StreamableMessageSource */
                                                          )
                                                          // ... to provide additional configuration ...
                                                          .registerTrackingEventProcessor(
                                                                  "my-processor",
                                                                  conf -> /* create/return StreamableMessageSource */,
                                                                  conf -> tepConfig
                                                          )
        );
    }
}
A properties file allows the configuration of some fields on an Event Processor. Do note that the Java configuration provides more degrees of freedom.
axon.eventhandling.processors.my-processor.mode=tracking
axon.eventhandling.processors.my-processor.source=eventStore
If the name of an event processor contains periods ., use the map notation:
axon.eventhandling.processors[my.processor].mode=tracking
axon.eventhandling.processors[my.processor].source=eventStore
For more fine-grained control when configuring a Tracking Processor, the TrackingEventProcessorConfiguration can be used. When invoking the registerTrackingEventProcessor method, you can provide a tracking processor configuration object, or you can register the configuration instance explicitly:
Axon Configuration API
Spring Boot AutoConfiguration - Java
public class AxonConfig {
    // omitting other configuration methods...
    public void registerTrackingProcessorConfig(EventProcessingConfigurer processingConfigurer) {
        TrackingEventProcessorConfiguration tepConfig =
                TrackingEventProcessorConfiguration.forSingleThreadedProcessing();
            
        // To register a default tracking config ...
        processingConfigurer.registerTrackingEventProcessorConfiguration(config -> tepConfig)
                            // ... to register a config for a specific processor.
                            .registerTrackingEventProcessorConfiguration("my-processor", config -> tepConfig);
    }
}
@Configuration
public class AxonConfig {
    // omitting other configuration methods...
    @Bean
    public ConfigurerModule trackingProcessorConfigurerModule() {
        TrackingEventProcessorConfiguration tepConfig =
              TrackingEventProcessorConfiguration.forSingleThreadedProcessing();
​
        return configurer -> configurer.eventProcessing(
              // To register a default tracking config ...
              processingConfigurer -> processingConfigurer.registerTrackingEventProcessorConfiguration(config -> tepConfig)
                                                          // ... to register a config for a specific processor.
                                                          .registerTrackingEventProcessorConfiguration(
                                                                  "my-processor", config -> tepConfig
                                                          )
        );
    }
}
Configuring a Pooled Streaming Processor
Firstly, to specify that every new processors should default to a PooledStreamingEventProcessor, you can invoke the usingPooledStreamingEventProcessors method:
Axon Configuration API
Spring Boot AutoConfiguration
public class AxonConfig { 
    // omitting other configuration methods...
    public void configureProcessorDefault(EventProcessingConfigurer processingConfigurer) { 
        processingConfigurer.usingPooledStreamingEventProcessors();  
    }
}
@Configuration
public class AxonConfig {
    // omitting other configuration methods...
    @Bean
    public ConfigurerModule processorDefaultConfigurerModule() {
        return configurer -> configurer.eventProcessing(EventProcessingConfigurer::usingPooledStreamingEventProcessors);
    }
}
For a specific Event Processor to be a Pooled Streaming instance, registerPooledStreamingProcessor is used:
Axon Configuration API
Spring Boot AutoConfiguration - Java
Spring Boot AutoConfiguration - Properties file
public class AxonConfig {
    // omitting other configuration methods...
    public void configurePooledStreamingProcessors(EventProcessingConfigurer processingConfigurer) {
          // This configuration object allows for fine-grained control over the Pooled Streaming Processor
        EventProcessingConfigurer.PooledStreamingProcessorConfiguration psepConfig = 
                (config, builder) -> builder/* ... */;
          
        // To configure a processor to be pooled streaming ...
        processingConfigurer.registerPooledStreamingEventProcessor("my-processor")
                            // ... to define a specific StreamableMessageSource ... 
                            .registerPooledStreamingEventProcessor(
                                    "my-processor", conf -> /* create/return StreamableMessageSource */
                            )
                            // ... to provide additional configuration ...
                            .registerPooledStreamingEventProcessor(
                                    "my-processor", conf -> /* create/return StreamableMessageSource */, psepConfig
                            );
    }
}
@Configuration
public class AxonConfig {
    // omitting other configuration methods...
    @Bean
    public ConfigurerModule pooledStreamingProcessorConfigurerModule() {
        // This configuration object allows for fine-grained control over the Pooled Streaming Processor
        EventProcessingConfigurer.PooledStreamingProcessorConfiguration psepConfig =
              (config, builder) -> builder/* ... */;
​
        return configurer -> configurer.eventProcessing(
              // To configure a processor to be pooled streaming ...
              processingConfigurer -> processingConfigurer.registerPooledStreamingEventProcessor("my-processor")
                                                          // ... to define a specific StreamableMessageSource ... 
                                                          .registerPooledStreamingEventProcessor(
                                                                  "my-processor",
                                                                  conf -> /* create/return StreamableMessageSource */
                                                          )
                                                          // ... to provide additional configuration ...
                                                          .registerPooledStreamingEventProcessor(
                                                                  "my-processor",
                                                                  conf -> /* create/return StreamableMessageSource */,
                                                                  psepConfig
                                                          )
        );
    }
}
A properties file allows the configuration of some fields on an Event Processor. Do note that the Java configuration provides more degrees of freedom.
axon.eventhandling.processors.my-processor.mode=pooled
axon.eventhandling.processors.my-processor.source=eventStore
If the name of an event processor contains periods ., use the map notation:
axon.eventhandling.processors[my.processor].mode=pooled
axon.eventhandling.processors[my.processor].source=eventStore
For more fine-grained control when configuring a Pooled Streaming Processor, the PooledStreamingProcessorConfiguration can be used. When invoking the registerPooledStreamingEventProcessor method, you can provide a pooled streaming processor configuration object, or you can register the configuration instance explicitly:
Axon Configuration API
Spring Boot AutoConfiguration - Java
public class AxonConfig {
    // omitting other configuration methods...
    public void registerPooledStreamingProcessorConfig(EventProcessingConfigurer processingConfigurer) {
        EventProcessingConfigurer.PooledStreamingProcessorConfiguration psepConfig = 
                (config, builder) -> builder/* ... */;
          
        // To register a default pooled streaming config ...
        processingConfigurer.registerPooledStreamingEventProcessorConfiguration(psepConfig)
                            // ... to register a config for a specific processor.
                            .registerPooledStreamingEventProcessorConfiguration("my-processor", psepConfig);
    }
}
@Configuration
public class AxonConfig {
    // omitting other configuration methods...
    @Bean
    public ConfigurerModule pooledStreamingProcessorConfigurerModule() {
        EventProcessingConfigurer.PooledStreamingProcessorConfiguration psepConfig =
              (config, builder) -> builder/* ... */;
​
        return configurer -> configurer.eventProcessing(
              // To register a default pooled streaming config ...
              processingConfigurer -> processingConfigurer.registerPooledStreamingEventProcessorConfiguration(psepConfig)
                                                          // ... to register a config for a specific processor.
                                                          .registerPooledStreamingEventProcessorConfiguration(
                                                                  "my-processor", psepConfig
                                                          )
        );
    }
}
Error Mode
The error mode differs between the Tracking- and Pooled Streaming Event Processor.
Whenever the error handler rethrows an exception, a TrackingEventProcessor will retry processing the event using an incremental back-off period. It will start at 1 second and double after each attempt until it reaches the maximum wait time of 60 seconds per attempt. This back-off time ensures that in a distributed environment, when another node is able to process events, it will have the opportunity to claim the token required to process the event.
The PooledStreamingEventProcessor simply aborts the failed part of the process. The Pooled Streaming Processor can deal with this since the threading mode is different from the Tracking Processor. As such, the chance is high the failed process will be picked up quickly by another thread within the same JVM. This chance increases further whenever the PSEP instance is distributed over several application instances.
Tracking Tokens
A vital attribute of the Streaming Event Processor is its capability to keep and maintain the processing progress. It does so through the TrackingToken, the "token" for short. Such a token accompanies each message a streaming processor receives through its event stream. It's this token that:
1.specifies the position of the event on the overall stream, and
2.is used by the Streaming Processor to open the event stream at the desired position on start-up.
Using tokens gives the Streaming Event Processor several benefits, like:
Being able to reopen the stream at any later point, picking up where it left off with the last event.
Dealing with unintended shutdowns without losing track of the last events they've handled.
Collaboration over the event handling load from two perspectives. First, the tokens make sure only a single thread is actively processing specific events. Secondly, it allows parallelization of the load over several threads or nodes of a Streaming Processor.
​Replaying events by adjusting the token position of that processor.
To be able to reopen the stream at a later point, we should keep the progress somewhere. The progress is kept by updating and saving the TrackingToken after handling batches of events. Keeping the progress requires CRUD operation, for which the Streaming Processor uses the TokenStore.
For a Streaming Processor to process any events, it needs "a claim" on a TrackingToken. The processor will update this claim every time it has finished handling a batch of events. This so-called "claim extension" is, just as updating and saving of tokens, delegated to the Token Store. Hence, the Streaming Processors achieves collaboration among instances/threads through token claims.
In the absence of a claim, a processor will actively try to retrieve one. If a token claim is not extended for a configurable amount of time, other processor threads can "steal" the claim. Token stealing can, for example, happen if event processing is slow or encountered some exceptions.
Initial Tracking Token
The Streaming Processor uses a StreamableMessageSource to retrieve a stream of events that will open on start-up. It requires a TrackingToken to open this stream, which it will fetch from the TokenStore. However, if a Streaming Processor starts for the first time, there is no TrackingToken present to open the stream with yet.
Whenever this situation occurs, a Streaming Processor will construct an "initial token." By default, the initial token will start at the tail of the event stream. Thus, the processor will begin at the start and handle every event present in the message source. This start position is configurable, as is described here.
A Saga's Streaming Processor initial position
A Streaming Processor dedicated to a Saga will default the initial token to the head of the stream. The default initial token position ensures that the Saga does not react to events from the past, as in most cases, this would introduce unwanted side effects.
Conceptually there are a couple of scenarios when a processor builds an initial token on application startup. The obvious one is already shared, namely when a processor starts for the first time. There are, however, also other situations when a token is built that might be unexpected, like:
The TokenStore has (accidentally) been cleared between application runs, thus losing the stored tokens.
The application running the processor starts in a new environment (e.g., test or acceptance) for the first time.
An InMemoryTokenStore was used, and hence the processor could never persist the token to begin with.
The application is (accidentally) pointing to another storage solution than expected.
Whenever a Streaming Processor's event handlers show unexpected behavior in the form of missed or reprocessed events, a new initial token might have been triggered. In those cases, we recommend to validate if any of the above situations occurred.
Token Configuration
There are a couple of things we can configure when it comes to tokens. We can separate these options in "initial token" and "token claim" configuration, as described in the following sections:
Initial Token
The initial token for a StreamingEventProcessor is configurable for every processor instance. When configuring the initial token builder function, the received input parameter is the StreamableMessageSource. The message source, in turn, gives three possibilities to build a token, namely:
1.createHeadToken() - Creates a token from the head of the event stream.
2.createTailToken() - Creates a token from the tail of the event stream. Creating tail tokens is the default value for most Streaming Processors.
3.createTokenAt(Instant) / createTokenSince(Duration) - Creates a token that tracks all events after a given time. If there is an event precisely at that given moment in time, it will also be taken into account.
Of course, you can completely disregard the StreamableMessageSource input parameter and create a token by yourself. Consider the following snippets if you want to configure a different initial token:
Tracking Processor - Axon Configuration API
Tracking Processor - Spring Boot AutoConfiguration
Pooled Streaming Processor - Axon Configuration API
Pooled Streaming Processor - Spring Boot AutoConfiguration
public class AxonConfig {
    // omitting other configuration methods...
    public void configureInitialTrackingToken(EventProcessingConfigurer processingConfigurer) {
        TrackingEventProcessorConfiguration tepConfig = 
                TrackingEventProcessorConfiguration.forSingleThreadedProcessing()
                                                   .andInitialTrackingToken(StreamableMessageSource::createHeadToken);
        
        processingConfigurer.registerTrackingEventProcessorConfiguration("my-processor", config -> tepConfig);
    }
}
@Configuration
public class AxonConfig {
    // omitting other configuration methods...
    @Bean
    public ConfigurerModule initialTrackingTokenConfigurerModule() {
        TrackingEventProcessorConfiguration tepConfig =
              TrackingEventProcessorConfiguration.forSingleThreadedProcessing()
                                                 .andInitialTrackingToken(StreamableMessageSource::createTailToken);
​
        return configurer -> configurer.eventProcessing(
              processingConfigurer -> processingConfigurer.registerTrackingEventProcessorConfiguration(
                      "my-processor", config -> tepConfig
              )
        );
    }
}
public class AxonConfig {
    // omitting other configuration methods...
    public void configureInitialTrackingToken(EventProcessingConfigurer processingConfigurer) {
        EventProcessingConfigurer.PooledStreamingProcessorConfiguration psepConfig = 
                (config, builder) -> builder.initialToken(messageSource -> messageSource.createTokenSince(
                        messageSource -> messageSource.createTokenAt(Instant.parse("20020-12-01T10:15:30.00Z"))
                ));
        
        processingConfigurer.registerPooledStreamingEventProcessorConfiguration("my-processor", psepConfig);
    }
}
@Configuration
public class AxonConfig {
    // omitting other configuration methods...
    @Bean
    public ConfigurerModule initialTrackingTokenConfigurerModule() {
        EventProcessingConfigurer.PooledStreamingProcessorConfiguration psepConfig =
              (config, builder) -> builder.initialToken(
                      messageSource -> messageSource.createTokenSince(Duration.ofDays(31))
              );
​
        return configurer -> configurer.eventProcessing(
              processingConfigurer -> processingConfigurer.registerPooledStreamingEventProcessorConfiguration(
                      "my-processor", psepConfig
              )
        );
    }
}
Token Claims
As described here, a streaming processor should claim a token before it is allowed to perform any processing work. There are several scenarios where a processor may keep the claim for too long. This can occur when, for example, the event handling process is slow or encountered an exception.
In those scenarios, another processor can steal a token claim to proceed with processing. There are a couple of configurable values that influence this process:
tokenClaimInterval - Defines how long to wait between attempts to claim a segment. A processor uses this value to steal token claims from other processor threads. This value defaults to 5000 milliseconds.
eventAvailabilityTimeout - Defines the time to "wait for events" before extending the claim. Only the Tracking Event Processor uses this. The value defaults to 1000 milliseconds.
claimExtensionThreshold - Threshold to extend the claim in the absence of events. Only the Pooled Streaming Event Processor uses this. The value defaults 5000 milliseconds.
Consider the following snippets if you want to configure any of these values:
Tracking Processor - Axon Configuration API
Tracking Processor - Spring Boot AutoConfiguration
Pooled Streaming Processor - Axon Configuration API
Pooled Streaming Processor - Spring Boot AutoConfiguration
public class AxonConfig {
    // omitting other configuration methods...
    public void configureTokenClaimValues(EventProcessingConfigurer processingConfigurer) {
        TrackingEventProcessorConfiguration tepConfig = 
                TrackingEventProcessorConfiguration.forSingleThreadedProcessing()
                                                   .andTokenClaimInterval(1000, TimeUnit.MILLISECONDS)
                                                   .andEventAvailabilityTimeout(2000, TimeUnit.MILLISECONDS);
        
        processingConfigurer.registerTrackingEventProcessorConfiguration("my-processor", config -> tepConfig);
    }
}
@Configuration
public class AxonConfig {
    // omitting other configuration methods...
    @Bean
    public ConfigurerModule tokenClaimValuesConfigurerModule() {
        TrackingEventProcessorConfiguration tepConfig =
              TrackingEventProcessorConfiguration.forSingleThreadedProcessing()
                                                 .andTokenClaimInterval(1000, TimeUnit.MILLISECONDS)
                                                 .andEventAvailabilityTimeout(2000, TimeUnit.MILLISECONDS);
​
        return configurer -> configurer.eventProcessing(
              processingConfigurer -> processingConfigurer.registerTrackingEventProcessorConfiguration(
                      "my-processor", config -> tepConfig
              )
        );
    }
}
public class AxonConfig {
    // omitting other configuration methods...
    public void configureTokenClaimValues(EventProcessingConfigurer processingConfigurer) {
        EventProcessingConfigurer.PooledStreamingProcessorConfiguration psepConfig = 
                (config, builder) -> builder.tokenClaimInterval(2000)
                                            .claimExtensionThreshold(3000);
        
        processingConfigurer.registerPooledStreamingEventProcessorConfiguration("my-processor", psepConfig);
    }
}
@Configuration
public class AxonConfig {
    // omitting other configuration methods...
    @Bean
    public ConfigurerModule tokenClaimValuesConfigurerModule() {
        EventProcessingConfigurer.PooledStreamingProcessorConfiguration psepConfig =
              (config, builder) -> builder.tokenClaimInterval(2000)
                                          .claimExtensionThreshold(3000);
​
        return configurer -> configurer.eventProcessing(
              processingConfigurer -> processingConfigurer.registerPooledStreamingEventProcessorConfiguration(
                      "my-processor", psepConfig
              )
        );
    }
}
Token Stealing
As described at the start, streaming processor threads can "steal" tokens from one another. A token is "stolen" when a thread loses a token claim. Situations like this internally result in an UnableToClaimTokenException, caught by both streaming event processor implementations and translated into warn- or info-level log statements.
Where the framework uses token claims to ensure that a single thread is processing a sequence of events, it supports token stealing to guarantee event processing is not blocked forever. In short, the framework uses token stealing to unblock your streaming processor threads when processing takes too long. Examples may include literal slow processing, blocking exceptional scenarios, and deadlocks.
However, token stealing may occur as a surprise for some applications, making it an unwanted side effect. As such, it is good to be aware of why tokens get stolen (as described above), but also when this happens and what the consequences are.
When is a Token stolen?
In practical terms, a token is stolen whenever the claim timeout is exceeded.
This timeout is met whenever the token's timestamp (e.g., the timestamp column of your token_entry table) exceeds the claimTimeout of the TokenStore. By default, the claimTimeout value equals 10 seconds. To adjust it, you must configure a TokenStore instance through its builder, as shown in the Token Store section.
The token's timestamp is equally crucial in deciding when the timeout is met. The streaming processor thread holding the claim is in charge of updating the token timestamp. This timestamp is updated whenever the thread finishes a batch of events or whenever the processor extends the claim. When to extend a claim differs between the Tracking and Pooled Streaming processor. You should check out the token claim section if you want to know how to configure these values.
To further clarify, a streaming processor's thread needs to be able to update the token claim and, by extension, the timestamp to ensure it won't get stolen. Hence, a staling processor thread will, one way or another, eventually lose the claim.
Examples of when a thread may get its token stolen are:
Overall slow event handling
Too large event batch size
Blocking operations inside event handlers
Blocking exceptions inside event handlers
What are the consequences of Token stealing?
The consequence of token stealing is that an event may be handled twice (or more).
When a thread steals a token, the original thread was already processing events from the token's position. To protect against doubling event handling, Axon Framework will combine committing the event handling task with updating the token. As the token claim is required to update the token, the original thread will fail the update. Following this, a rollback occurs on the Unit of Work, resolving most issues arising from token stealing.
The ability to rollback event handling tasks sheds light on the consequences of token stealing. Most event processors project events into a projection stored within a database. Furthermore, if you store the projection in the same database as the token, the rollback will ensure the change is not persisted. Thus, the consequence of token stealing is limited to wasting processor cycles. This scenario is why we recommend storing tokens and projections in the same database.
If a rollback is out of the question for an event handling task, we strongly recommend making the task idempotent. You may have this scenario when, for example, the projection and tokens do not reside in the same database. or when the event handler dispatches an operation (e.g., through the CommandGateway). In making the invoked operation idempotent, you ensure that whenever the thread stealing a token handles an event twice (or more), the outcome will be identical.
Without idempotency, the consequences of token stealing can be manyfold:
Your projection (stored in a different database than your tokens!) may incorrectly project the state.
An event handler putting messages on a queue will put a message on the queue again.
A Saga Event Handler invoking a third-party service will invoke that service again.
An event handler sending an email will send that email again.
In short, any operation introducing a side effect that isn't handled in an idempotent fashion will occur again when a token is stolen.
Concluding, we can separate the consequence of token stealing into roughly three scenarios:
1.We can rollback the operation. In this case, the only consequence is wasted processor cycles.
2.The operation is idempotent. In this case, the only consequence is wasted processor cycles.
3.When the task cannot be rolled back nor performed in an idempotent fashion, compensating actions may be the way out.
Token Store
The TokenStore provides the CRUD operations for the StreamingEventProcessor to interact with TrackingTokens. The streaming processor will use the store to construct, fetch and claim tokens.
When no token store is explicitly defined, an InMemoryTokenStore is used. The InMemoryTokenStore is not recommended in most production scenarios since it cannot maintain the progress through application shutdowns. Unintentionally using the InMemoryTokenStore counts towards one of the unexpected scenarios where the framework creates an initial token on each application start-up.
The framework provides a couple of TokenStore implementations:
InMemoryTokenStore - A TokenStore implementation that keeps the tokens in memory. This implementation does not suffice as a production-ready store in most applications.
JpaTokenStore - A TokenStore implementation using JPA to store the tokens with. Expects that a table is constructed based on the org.axonframework.eventhandling.tokenstore.jpa.TokenEntry. It is easily auto-configurable with, for example, Spring Boot.
JdbcTokenStore - A TokenStore implementation using JDBC to store the tokens with. Expects that the schema is constructed through the JdbcTokenStore#createSchema(TokenTableFactory) method. Several TokenTableFactory can be chosen here, like the GenericTokenTableFactory, PostgresTokenTableFactory or Oracle11TokenTableFactory implementation.
MongoTokenStore- A TokenStore implementation using Mongo to store the tokens with.
Where to store Tokens?
Where possible, we recommend using a token store that stores tokens in the same database as to where the event handlers update the view models. This way, changes to the view model can be stored atomically with the changed tokens. Furthermore, it guarantees exactly-once processing semantics.
Note that you can configure the token store to use for a streaming processor in the EventProcessingConfigurer:
Axon Configuration API
Spring Boot AutoConfiguration
To configure a TokenStore for all processors:
public class AxonConfig { 
    // omitting other configuration methods...
    public void registerTokenStore(EventProcessingConfigurer processingConfigurer) {
        TokenStore tokenStore = JpaTokenStore.builder()
                                             // …
                                             .build();
    
        processingConfigurer.registerTokenStore(config -> tokenStore);
    }
}
Alternatively, to configure a TokenStore for a specific processor, use:
public class AxonConfig { 
    // omitting other configuration methods...
    public void registerTokenStore(EventProcessingConfigurer processingConfigurer, String processorName) {
        TokenStore tokenStore = JdbcTokenStore.builder()
                                              // …
                                              .build();
    
        processingConfigurer.registerTokenStore(processorName, config -> tokenStore);
    }
}
The default TokenStore implementation is defined based on dependencies available in Spring Boot, in the following order:
1.If any TokenStore bean is defined, that bean is used.
2.Otherwise, if an EntityManager is available, the JpaTokenStore is defined.
3.Otherwise, if a DataSource is defined, the JdbcTokenStore is created.
4.Lastly, the InMemoryToken store is used.
To override the TokenStore, either define a bean in a Spring @Configuration class:
@Configuration
public class AxonConfig {
    // omitting other configuration methods...
    @Bean
    public TokenStore myTokenStore() {
        return JpaTokenStore.builder()
                            // …
                            .build();
    }
}
Alternatively, inject the EventProcessingConfigurer, which allows more fine-grained customization:
@Configuration
public class AxonConfig {
    // omitting other configuration methods...
    @Bean
    public ConfigurerModule tokenStoreConfigurerModule() {
        TokenStore tokenStore = JdbcTokenStore.builder()
                                              // …
                                              .build();
​
        return configurer -> configurer.eventProcessing(
              processingConfigurer -> processingConfigurer.registerTokenStore(conf -> tokenStore)
                                                          // or, to define one for a specific processor:
                                                          .registerTokenStore("my-processor", conf -> tokenStore)
        );
    }
}
Retrieving the Token Store Identifier
Implementations of TokenStore might share state in the underlying storage. To ensure correct operation, a token store has a unique identifier that uniquely identifies the storage location of the tokens in that store. This identifier can be queried with the retrieveStorageIdentifier method of your event processor.
StreamingEventProcessor eventProcessor = // …
String tokenStoreId =  eventProcessor.getTokenStoreIdentifier();
Parallel Processing
Streaming processors can use multiple threads to process an event stream. Using multiple threads allows the StreamingEventProcessor to more efficiently process batches of events. As described here, a streaming processor's thread requires a claim on a tracking token to process events.
Thus, to be able to parallelize the load, we require several tokens per processor. To that end, each token instance represents a segment of the event stream, wherein each segment is identified through a number. The stream segmentation approach ensures events aren't handled twice (or more), as that would otherwise introduce unintentional duplication. Due to this, the Streaming Processor's API references segment claims instead of token claims throughout.
You can define the number of segments used by adjusting the initialSegmentCount property. Only when a streaming processor starts for the first time can it initialize the number of segments to use. This requirement follows from the fact each token represents a single segment. Tokens, in turn, can only be initialized if they are not present yet, as is explained in more detail here.
Whenever the number of segments should be adjusted during runtime, you can use the split and merge functionality. To adjust the number of initial segments, consider the following sample:
Tracking Processor - Axon Configuration API
Tracking Processor - Spring Boot AutoConfiguration
Pooled Streaming Processor - Axon Configuration API
Pooled Streaming Processor - Spring Boot AutoConfiguration
Spring Boot AutoConfiguration - Properties File
The default number of segments for a TrackingEventProcessor is one.
public class AxonConfig {
    // omitting other configuration methods...
    public void configureSegmentCount(EventProcessingConfigurer processingConfigurer) {
        TrackingEventProcessorConfiguration tepConfig = 
                TrackingEventProcessorConfiguration.forParallelProcessing(2)
                                                   .andInitialSegmentsCount(2);
        
        processingConfigurer.registerTrackingEventProcessorConfiguration("my-processor", config -> tepConfig);
    }
}
The default number of segments for a TrackingEventProcessor is one.
@Configuration
public class AxonConfig {
    // omitting other configuration methods...
    @Bean
    public ConfigurerModule segmentCountConfigurerModule() {
        TrackingEventProcessorConfiguration tepConfig =
              TrackingEventProcessorConfiguration.forParallelProcessing(2)
                                                 .andInitialSegmentsCount(2);
​
        return configurer -> configurer.eventProcessing(
              processingConfigurer -> processingConfigurer.registerTrackingEventProcessorConfiguration(
                      "my-processor", config -> tepConfig
              )
        );
    }
}
The default number of segments for a PooledStreamingEventProcessor is sixteen.
public class AxonConfig {
    // omitting other configuration methods...
    public void configureSegmentCount(EventProcessingConfigurer processingConfigurer) {
        EventProcessingConfigurer.PooledStreamingProcessorConfiguration psepConfig =
                (config, builder) -> builder.initialSegmentCount(32);
        
        processingConfigurer.registerPooledStreamingEventProcessorConfiguration("my-processor", psepConfig);
    }
}
The default number of segments for a PooledStreamingEventProcessor is sixteen.
@Configuration
public class AxonConfig {
    // omitting other configuration methods...
    @Bean
    public ConfigurerModule segmentCountConfigurerModule() {
        EventProcessingConfigurer.PooledStreamingProcessorConfiguration psepConfig =
              (config, builder) -> builder.initialSegmentCount(32);
​
        return configurer -> configurer.eventProcessing(
              processingConfigurer -> processingConfigurer.registerPooledStreamingEventProcessorConfiguration(
                      "my-processor", psepConfig
              )
        );
    }
}
The default number of segments for a TrackingEventProcessor and PooledStreamingEventProcessor is one and sixteen, respectively.
axon.eventhandling.processors.my-processor.mode=pooled
# Sets the initial number of segments
axon.eventhandling.processors.my-processor.initialSegmentCount=32
Parallel Processing and Subscribing Event Processors
Note that Subscribing Event Processor don't manage their own threads. Therefore, it is not possible to configure how they should receive their events. Effectively, they will always work on a sequential-per-aggregate basis, as that is generally the level of concurrency in the command handling component.
The Event Handling Components a processor is in charge of may have specific expectations on the event order. The ordering is guaranteed when only a single thread is processing events. Maintaining the ordering requires additional work when the stream is segmented for parallel processing, however. When this is the case, the processor must ensure it sends the events to these handlers in that specific order.
Axon uses the SequencingPolicy for this. The SequencingPolicy is a function that returns a value for any given message. If the return value of the SequencingPolicy function is equal for two distinct event messages, it means that those messages must be processed sequentially. By default, Axon components will use the SequentialPerAggregatePolicy, making it so that events published by the same aggregate instance will be handled sequentially. Check out this section to understand how to influence the sequencing policy.
Each node running a streaming processor will attempt to start its configured amount of threads to start processing events. The number of segments that a single thread can claim differ between the Tracking- and Pooled Streaming Event Processor. A tracking processor can only claim a single segment per thread, whereas the pooled streaming processor can claim any amount of segments per thread. These approaches provide different pros and cons for each implementation, which this section explains further.
Sequential Processing
Even though events are processed asynchronously from their publisher, it is often desirable to process certain events in their publishing order. In Axon, the SequencingPolicy controls this order. The SequencingPolicy defines whether events must be handled sequentially, in parallel, or a combination of both. Policies return a sequence identifier of a given event.
If the policy returns the same identifier for two events, they must be handled sequentially by the Event Handling Component. Thus, if the SequencingPolicy returns a different value for two events, they may be processed concurrently. Note that if the policy returns a null sequence identifier, the event may be processed in parallel with any other events.
** Parallel Processing and Sagas**
A saga instance is never invoked concurrently by multiple threads. Therefore, the SequencingPolicy is irrelevant for a saga. Axon will ensure each saga instance receives the events it needs to process in the order they have been published on the event bus.
Conceptually, the SequencingPolicy decides whether an event belongs to a given segment. Furthermore, Axon guarantees that Events that are part of the same segment are processed sequentially.
The framework provides several policies you can use out of the box:
SequentialPerAggregatePolicy - The default policy. It will force domain events that were raised from the same aggregate to be handled sequentially. Thus, events from different aggregates may be handled concurrently. This policy is typically suitable for Event Handling Components that update details from aggregates in databases.
FullConcurrencyPolicy - This policy will tell Axon that this Event Processor may handle all events concurrently. This means that there is no relationship between the events that require them to be processed in a particular order.
SequentialPolicy - This policy tells Axon that it can process all events sequentially. Handling of an event will start when the handling of a previous event has finished.
PropertySequencingPolicy - When configuring this policy, the user is required to provide a property name or property extractor function. This implementation provides a flexible solution to set up a custom sequencing policy based on a standard value present in your events. Note that this policy only reacts to properties present in the event class.
MetaDataSequencingPolicy - When configuring this policy, the user is required to provide a metaDataKey to be used. This implementation provides a flexible solution to set up a custom sequencing policy based on a standard value present in your events' metadata.
Consider the following snippets when configuring a (custom) SequencingPolicy:
Axon Configuration API
Spring Boot AutoConfiguration
Spring Boot AutoConfiguration - Properties File
public class AxonConfig {
    // omitting other configuration methods...
    public void configureSequencingPolicy(EventProcessingConfigurer processingConfigurer) {
          PropertySequencingPolicy<SomeEvent, String> mySequencingPolicy = 
                  PropertySequencingPolicy.builder(SomeEvent.class, String.class)
                                          .propertyName("myProperty")
                                          .build();
          
          processingConfigurer.registerDefaultSequencingPolicy(config -> mySequencingPolicy)
                              // or, to define one for a specific processor:
                              .registerSequencingPolicy("my-processor", config -> mySequencingPolicy);
    }
}
@Configuration
public class AxonConfig {
    // omitting other configuration methods...
    @Bean
    public ConfigurerModule sequencingPolicyConfigurerModule(SequencingPolicy<EventMessage<?>> mySequencingPolicy) {
        return configurer -> configurer.eventProcessing(
              processingConfigurer -> processingConfigurer.registerDefaultSequencingPolicy(config -> mySequencingPolicy)
                                                          // or, to define one for a specific processor:
                                                          .registerSequencingPolicy("my-processor", config -> mySequencingPolicy)
        );
    }
​
    @Bean
    public SequencingPolicy<EventMessage<?>> mySequencingPolicy() {
        return new SequentialPolicy();
    }
}
When we want to configure the SequencingPolicy in a properties file, we should provide a bean name:
axon.eventhandling.processors.my-processor.mode=tracking
axon.eventhandling.processors.my-processor.sequencing-policy=mySequencingPolicy
This approach does require the bean name to be present in the Application Context of course:
@Configuration
public class AxonConfig {
    // omitting other configuration methods...
    @Bean
    public SequencingPolicy<EventMessage<?>> mySequencingPolicy() {
        return new FullConcurrencyPolicy();
    }
}
If the available policies do not suffice, you can define your own. To that end, we should implement the SequencingPolicy interface. This interface defines a single method, getSequenceIdentifierFor(T), that returns the sequence identifier for a given event:
public interface SequencingPolicy<T> {
    
    Object getSequenceIdentifierFor(T event);
}
Thread Configuration
A Streaming Processor cannot process events in parallel without multiple threads configured. We can process events in parallel by running several nodes of an application. Or by configuring a StreamingEventProcessor to use several threads. The following section describes the threading differences between the Tracking- and Pooled Streaming Event Processor. These sections are followed up with samples on configuring multiple threads for the TEP and PSEP, respectively.
Thread and Segment Count
Adjusting the number of threads will not automatically parallelize a Streaming Processor. A segment claim is required to let a thread process any events. Hence, increasing the thread count should be paired with adjusting the segment count.
Tracking Processor Threading
The TrackingEventProcessor uses a ThreadFactory to start the process of claiming segments. It will use a single thread per segment it is able to claim until the processor exhausts the configured amount of threads. Each thread will open a stream with the StreamableMessageSource and start processing events at their own speed. Other segment operations, like split and merge, are processed by the thread owning the segment operated on.
Since the tracking processor can only claim a single segment per thread, segments may go unprocessed if there are more segments than threads. Hence, we recommend setting the number of threads (on every node) higher than or equal to the total number of segments.
To increase event handling throughput, we recommend changing the number of threads. How to do this is shown in the following sample:
Axon Configuration API
Spring Boot AutoConfiguration
Spring Boot AutoConfiguration - Properties File
public class AxonConfig {
    // omitting other configuration methods...
    public void configureThreadCount(EventProcessingConfigurer processingConfigurer) {
        TrackingEventProcessorConfiguration tepConfig =
                TrackingEventProcessorConfiguration.forParallelProcessing(4)
                                                   .andInitialSegmentsCount(4);
​
        processingConfigurer.registerTrackingEventProcessorConfiguration("my-processor", config -> tepConfig);
    }
}
@Configuration
public class AxonConfig {
    // omitting other configuration methods...
    @Bean
    public ConfigurerModule threadCountConfigurerModule() {
        TrackingEventProcessorConfiguration tepConfig =
              TrackingEventProcessorConfiguration.forParallelProcessing(4)
                                                 .andInitialSegmentsCount(4);
​
        return configurer -> configurer.eventProcessing(
              processingConfigurer -> processingConfigurer.registerTrackingEventProcessorConfiguration(
                      "my-processor", config -> tepConfig
              )
        );
    }
}
axon.eventhandling.processors.my-processor.mode=tracking
axon.eventhandling.processors.my-processor.thread-count=4
axon.eventhandling.processors.my-processor.initial-segment-count=4
Pooled Streaming Processor Threading
The PooledStreamingEventProcessor uses two threads pools instead of the single fixed set of threads used by the TrackingEventProcessor. The first thread pool is in charge of opening a stream with the event source, claiming as many segments as possible, and delegating all the work.
The work it coordinates is foremost the events to handle. Next to event coordination, it deals with segment operations like split and merge. The component coordinating all the work is called the Coordinator. This coordinator defaults to using a ScheduledExecutorService with a single thread, which suffices in most scenarios.
The second thread pool deals with all the segments the Coordinator of the pooled streaming processor could claim. The Coordinator starts a WorkPackage for each segment and provides them the events to handle. The work package will, in turn, invoke the Event Handling Components to process the events. These packages run within the second thread pool, the so-called "worker executor" pool. The worker-pool also defaults to ScheduledExecutorService with a single thread.
When you want to increase event handling throughput, we recommend changing the number of threads for the worker thread pool. How to do this is shown in the following sample:
Axon Configuration API
Spring Boot AutoConfiguration
Spring Boot AutoConfiguration - Properties File
public class AxonConfig {
    // omitting other configuration methods...
    public void configureThreadCount(EventProcessingConfigurer processingConfigurer) {
        // the "name" is the name of the processor, which can be used to define the thread factory name
        Function<String, ScheduledExecutorService> coordinatorExecutorBuilder =
                name -> Executors.newScheduledThreadPool(1, new AxonThreadFactory("Coordinator - " + name));
​
        Function<String, ScheduledExecutorService> workerExecutorBuilder =
                name -> Executors.newScheduledThreadPool(16, new AxonThreadFactory("Worker - " + name));
​
        EventProcessingConfigurer.PooledStreamingProcessorConfiguration psepConfig =
                (config, builder) -> builder.coordinatorExecutor(coordinatorExecutorBuilder)
                                            .workerExecutor(workerExecutorBuilder)
                                            .initialSegmentCount(32);
​
        processingConfigurer.registerPooledStreamingEventProcessorConfiguration("my-processor", psepConfig);
    }
}
@Configuration
public class AxonConfig {
    // omitting other configuration methods...
    @Bean
    public ConfigurerModule threadCountConfigurerModule() 