Spring Boot Configuration for Axon Framework with Tracking Token

A detailed guide on setting up tracking tokens for Axon event processors in a Spring Boot application

Key Takeaways

• Dependency and Auto-Configuration: Including the Axon Spring Boot Starter simplifies infrastructure setup and auto-configuration.
• Tracking Event Processor Setup: Precise configuration of tracking tokens ensures events are consumed correctly, either from the beginning, tail, or a specific point.
• Customizing and Persisting Tokens: Use a custom configuration class and a persistent TokenStore (e.g., JDBC) to manage and store tracking tokens reliably.

Introduction

Axon Framework empowers developers to build event-driven microservices and systems using Domain-Driven Design (DDD) and Command Query Responsibility Segregation (CQRS) patterns. When using Axon with Spring Boot, you benefit from auto-configuration that sets up key infrastructure components, including the Command Bus, Event Bus, and Query Bus. One of the advanced features of Axon is the tracking event processor, which uses a tracking token to keep a pointer on the event stream. This allows the processor to reliably resume event processing after interruptions. This guide will walk you through configuring a tracking token for an event processor in a Spring Boot application using Axon Framework.

Tracking tokens are used by the tracking event processors to remember their position in the event stream. In contrast to subscribing processors, tracking processors store their progress—commonly in a persistent token store—and can be restarted or scaled with minimal disruption. This approach is ideal for projects needing replay capabilities, load balancing, or handling events asynchronously.

Prerequisites and Dependencies

Before configuring the tracking token, ensure that you have included the Axon Spring Boot Starter dependency in your project. This adds the necessary Axon modules and integrates with Spring Boot automatically. If you're using Maven, your pom.xml should include:

<dependency>
    <groupId>org.axonframework</groupId>
    <artifactId>axon-spring-boot-starter</artifactId>
    <version>4.10</version>
</dependency>
  

For Gradle users, add the following to your build.gradle file:

implementation 'org.axonframework:axon-spring-boot-starter:4.10'
  

These dependencies will trigger Spring Boot auto-configuration and set up the fundamental components necessary for Axon to work seamlessly within your Spring Boot application.

Setting Up a Tracking Event Processor

One of the core components that can be customized is the tracking event processor. This processor utilizes a tracking token to manage its position in the event stream and to resume processing without reprocessing already handled events.

Java Configuration Approach

The following Java configuration class demonstrates how to configure a tracking event processor with a tracking token. In the example below, we use the Spring Boot auto-configuration and manually define a configuration that:

• Registers a persistent token store using JDBC.
• Sets up a single-threaded tracking processor that starts from either the head (beginning of the stream) or tail (only new events) using a token supplier.
• Assigns the processor to a specific processing group for the event handlers.

Example Configuration Class

// Import statements omitted for brevity
package com.example.config;

import org.axonframework.config.EventProcessingConfigurer;
import org.axonframework.eventhandling.tokenstore.jdbc.JdbcTokenStore;
import org.axonframework.eventhandling.TrackingEventProcessorConfiguration;
import org.axonframework.eventhandling.StreamableMessageSource;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import javax.sql.DataSource;

@Configuration
public class AxonConfig {

    /<b>
     * Configures a persistent TokenStore using JDBC.
     */
    @Bean
    public JdbcTokenStore tokenStore(DataSource dataSource) {
        return JdbcTokenStore.builder()
                             .dataSource(dataSource)
                             .build();
    }

    /</b>
     * Sets up the tracking event processor with a custom token supplier.
     * In this configuration, we showcase both starting from the beginning (head) of the event stream 
     * or using the tail token to process only new events.
     */
    @Bean
    public void configureTrackingProcessor(EventProcessingConfigurer configurer,
                                             JdbcTokenStore tokenStore,
                                             StreamableMessageSource<?> streamableMessageSource) {

        // Create a custom configuration for the tracking event processor.
        // Here, you can choose to initialize the tracking token as head or tail.
        TrackingEventProcessorConfiguration tepConfig = TrackingEventProcessorConfiguration
                .forSingleThreadedProcessing()
                // Use this if you want to start from the beginning of the event stream:
                //.andInitialTrackingToken(StreamableMessageSource::createHeadToken)
                // Alternatively, start from the tail to process only new events:
                .andInitialTrackingToken(streamableMessageSource::createTailToken);

        // Register a token store for the processor named "myEventProcessor"
        configurer.registerTokenStore("myEventProcessor", tokenStore);

        // Register the tracking event processor with the custom configuration
        configurer.registerTrackingEventProcessor(
                "myEventProcessor",
                configuration -> tepConfig,
                configuration -> streamableMessageSource
        );

        // Optionally, assign the processor to a processing group responsible for the relevant event handlers
        configurer.assignProcessingGroup("myEventProcessor", "myEventHandler");
    }
}
  

In this setup, the JdbcTokenStore persists tracking tokens in a relational database, ensuring that your processor’s progress is maintained even across application restarts. By leveraging a lambda function, the configuration sets the initial token either using createHeadToken or createTailToken based on your requirements.

Configuration via YAML Properties

In addition to Java configuration, you can configure tracking event processors in Axon Framework directly through your application.yml or application.properties file. This method is particularly useful for initial setup before proceeding to more advanced configurations.

Example YAML Configuration

axon:
  eventhandling:
    processors:
      myProcessorGroup:
        mode: tracking       # Set the processor mode to tracking
        threadCount: 2       # Optionally configure parallel processing (number of threads)
        source: eventStore   # Define the event source
  

The YAML configuration above instructs Axon to create a tracking processor for a group labeled myProcessorGroup, running in tracking mode. While these settings manage the basic behavior and resource allocation (like thread count), the token initialization strategy (whether to start processing events from the start or only new events) still needs to be managed within the Java configuration when overriding defaults is necessary.

Advanced Configuration Options

Beyond the basic setup, Axon Framework allows advanced configuration such as parallel processing, multiple segments, and starting from a specific point in time using a custom tracking token. This flexibility is essential when developing production-grade applications that might require high throughput or need to recover gracefully from downtime.

Parallel Processing and Custom Segments

For systems handling a large number of events, enabling parallel processing helps improve throughput. This can be achieved with the builder method forParallelProcessing as opposed to forSingleThreadedProcessing.

Here’s an example snippet demonstrating parallel processing with a configurable number of segments:

// In your configuration class
@Bean
public void configureParallelTrackingProcessor(EventProcessingConfigurer configurer,
                                                 StreamableMessageSource<?> streamableMessageSource) {
    TrackingEventProcessorConfiguration parallelConfig =
            TrackingEventProcessorConfiguration
                    .forParallelProcessing(2)       // Enable parallel processing with 2 threads
                    .andInitialSegmentsCount(1)       // Define the initial number of segments
                    .andInitialTrackingToken(streamableMessageSource::createHeadToken); // Start from the head
 
    configurer.registerTrackingEventProcessorConfiguration(
            "parallelProcessorGroup", 
            config -> parallelConfig
    );
}
  

In this configuration, the tracking event processor for the group parallelProcessorGroup is set up to process events using 2 threads. Adjusting the segment count allows you to fine-tune performance and balance load across available threads.

Starting from a Specific Point

There may be cases where you wish to replay events from a specific time or resume processing from a user-defined offset in the event stream. This requires setting the initial tracking token to a particular position. For instance, you can modify the token supplier to generate a token corresponding to a known event timestamp.

Example: Starting from a Specific Date and Time

import java.time.Instant;
// Inside your configuration class
public void configureInitialTokenWithTimestamp(EventProcessingConfigurer configurer,
                                                 StreamableMessageSource<?> messageSource) {
    TrackingEventProcessorConfiguration tokenConfig =
        TrackingEventProcessorConfiguration
            .forSingleThreadedProcessing()
            .andInitialTrackingToken(ts -> ts.createTokenSince(ts, Instant.parse("2020-12-01T10:15:30.00Z")));

    configurer.registerTrackingEventProcessorConfiguration("timestampBasedProcessor", c -> tokenConfig);
}
  

In this scenario, the processing group timestampBasedProcessor is initialized such that event consumption starts from December 1, 2020, at 10:15:30 UTC. This approach is useful for replaying historical events or resetting the event stream to a known snapshot.

Working with the TokenStore

A critical part of configuring a tracking event processor is ensuring the tracking token is stored persistently. The TokenStore implementation you use (such as JdbcTokenStore) guarantees that progress is maintained between application restarts, thereby preventing the reprocessing of events that have already been handled.

JDBC TokenStore

Using a JDBC-backed token store ensures that tokens are saved in your relational database. This is especially important in a production environment where consistency and fault tolerance are required. The following configuration snippet demonstrates how to set up a JDBC TokenStore:

@Bean
public JdbcTokenStore tokenStore(DataSource dataSource, org.axonframework.serialization.Serializer serializer) {
    return JdbcTokenStore.builder()
                         .dataSource(dataSource)
                         .serializer(serializer)
                         .build();
}
  

Once the JdbcTokenStore is set up, you can then register it with your tracking event processor configuration as shown previously. By doing so, the token used by the event processor is both managed and persisted, which is crucial for recovering from errors or scaling the application horizontally.

Integrating with Spring Data and Persistence Context

When using Spring Data JPA, Spring Boot auto-configuration takes care of setting up the persistence context. Axon ensures that its JPA entities (which include those for event sourcing and token storage) are integrated automatically within this context. This seamless integration helps reduce configuration overhead while still allowing you to customize as needed.

Comparison Table: Tracking Processor Configurations

The following table summarizes and compares various configurations you can apply to the tracking event processor:

This table provides an at-a-glance view of some of the key configuration options available when setting up a tracking event processor using Axon Framework.

Best Practices and Considerations

When configuring a tracking token for your event processor, here are some best practices to consider:

1. Consistent Naming Conventions

Consistently name your processing groups and event processors. This not only simplifies configuration management but also helps you easily identify which processor is handling which events. For example, clearly associate event handlers with processing groups in the configuration.

2. Decide on the Initial Token Strategy

Determine whether your processor should start at the head of the stream (processing historical events) or at the tail (processing only new events). If a replay of events is necessary—for example, after deploying a significant update—ensure your configuration can accommodate setting a custom token.

3. Monitor Token Health

It is crucial to monitor the health of your token store and ensure that tokens are not lost or corrupted. Employ logging and monitoring tools to observe the behavior of your tracking processors, and set up alerts for any anomalies in event consumption.

4. Use Persistent Storage

For production environments, always use a persistent TokenStore (like a JDBC TokenStore) to ensure that your tracking token persists across application restarts and deployments. This prevents reprocessing of events, which can lead to data inconsistencies or unintended side effects.

Conclusion

Setting up a Spring Boot application with Axon Framework to use tracking tokens for event processors offers a robust and flexible mechanism to manage event consumption. By including the Axon Spring Boot Starter dependency, both auto-configuration and custom configuration options become available. This guide detailed creating a custom configuration class to register tracking event processors, configuring your initial tracking token (choosing between the stream’s head or tail), and ensuring persistence using a JDBC-backed TokenStore. Advanced options like parallel processing, custom segmentation, and initializing a token from a specific point in time further empower your system to handle complex event processing scenarios.

Following these best practices and detailed configuration strategies not only enhances the reliability of your event processing capabilities but also allows for greater resilience in dynamic production environments. With careful planning and implementation, the integration of a tracking token in your Axon Framework setup provides significant benefits in scalability, maintainability, and overall system robustness.

References

• Axon Framework Streaming Processors - AxonIQ
• Axon Event Processors Overview - AxonIQ
• Axon Framework GitHub Repository - GitHub
• Axon Core Javadoc - Javadoc.io
• Axon Framework Discussions - Stack Overflow

More

How do I configure custom event processors in Axon Framework?

What is the role of a TokenStore in Axon Framework?

How can I enable parallel processing for event streams in Axon?