Projectors and Event Processing Migration

This path covers the migration of event processors and projection-related components (projectors) from Axon Framework 4.x to Axon Framework 5.

Message handler annotation changes

The annotations for handling events and queries have moved to new packages to align with the more explicit message-centric model of Axon Framework 5. For a detailed list of all import changes, refer to the Base import and package Changes section.

In particular, for message handlers this applies to:

Event Handler annotation: Moved from org.axonframework.eventhandling.EventHandler to org.axonframework.messaging.eventhandling.annotation.EventHandler
Query Handler annotation: Moved from org.axonframework.queryhandling.QueryHandler to org.axonframework.messaging.queryhandling.annotation.QueryHandler

Removal of `TrackingEventProcessor`

The TrackingEventProcessor has been removed in Axon Framework 5. Use the PooledStreamingEventProcessor as a direct replacement. The PooledStreamingEventProcessor is a more robust and performant implementation of the streaming processor concept.

For more information on how to configure and use the PooledStreamingEventProcessor, refer to the Streaming Event Processor section in the regular documentation.

Removal of `@ProcessingGroup`

The @ProcessingGroup annotation has been removed in Axon Framework 5. Previously, this annotation was used to group event handling components into a single event processor.

In Axon Framework 5, you can use the @Namespace annotation to assign an event handling component to an event processor.

// AF4
@ProcessingGroup("my-processor")
public class MyProjector {
    @EventHandler
    public void on(MyEvent event) { /* ... */ }
}

// AF5
import org.axonframework.messaging.core.annotation.Namespace;
import org.axonframework.messaging.eventhandling.annotation.EventHandler;

@Namespace("my-processor")
public class MyProjector {

    @EventHandler
    public void on(MyEvent event) {
        // ...
    }
}

Alternatively, you can use the declarative configuration to assign event handling components to processors.

Event processor configuration

The way event processors are configured has changed significantly to provide a more modular and explicit API.

Non-Spring users

For non-Spring users, the old EventProcessingConfigurer (and EventProcessingModule) has been replaced by a new flow accessed through the MessagingConfigurer#eventProcessing method.

// AF4
public void configure(Configurer configurer) {
    configurer.eventProcessing()
              .registerPooledStreamingEventProcessor("my-processor");
}

// AF5
import org.axonframework.messaging.core.configuration.MessagingConfigurer;

public class EventProcessorConfigurationExample {

    public void configure(MessagingConfigurer configurer) {
        configurer.eventProcessing(
                eventProcessing -> eventProcessing.pooledStreaming(
                        pooledStreaming -> pooledStreaming.processor(
                                "my-processor",
                                module -> module.eventHandlingComponents(
                                                         components -> components.autodetected(
                                                                 "my-projector",
                                                                 cfg -> new MyProjector()
                                                         )
                                                 )
                                                 .notCustomized()
                        )
                )
        );
    }
}

Spring users

For Spring users, the configuration of event processors has moved towards the Event Processor Specification (implemented via EventProcessorSettings). This replaces the previous way of configuring processors through the EventProcessingConfigurer bean or complex property structures.

Axon Framework 4
Axon Framework 5

import org.axonframework.config.ConfigurerModule;
import org.axonframework.config.EventProcessingConfigurer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class AxonConfig {

    @Bean
    public ConfigurerModule configure() {
        return configurer -> {
            EventProcessingConfigurer processingConfigurer = configurer.eventProcessing();
            processingConfigurer.registerPooledStreamingEventProcessor(
                                        "my-processor",
                                        org.axonframework.config.Configuration::eventStore,
                                        (config, builder) -> builder.initialSegmentCount(8)
                                                                    .batchSize(100)
                                )
                                .assignHandlerTypesMatching(
                                        "my-processor",
                                        type -> type.getPackageName().startsWith("com.my.projectors")
                                );
        };
    }
}

import org.axonframework.extension.spring.config.EventProcessorDefinition;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class AxonConfig {

    @Bean
    public EventProcessorDefinition myProcessorDefinition() {
        return EventProcessorDefinition.pooledStreaming("my-processor")
                                       .assigningHandlers(
                                               descriptor -> descriptor.beanType().getPackageName()
                                                                       .startsWith("com.my.projectors")
                                       )
                                       .customized(config -> config.initialSegmentCount(8)
                                                                   .batchSize(100));
    }
}

With Spring Boot you can also configure event processors using properties, which will be automatically converted to the appropriate EventProcessorSettings by Spring Boot’s configuration binding.

# AF5 Spring Boot properties example
axon:
  eventhandling:
    processors:
      my-processor:
        mode: pooled
        thread-count: 8

In your code, you can customize these settings by providing EventProcessorSettings beans or using the EventProcessorSettings.MapWrapper.

Initial token no longer marks the first pass as a replay

A streaming processor that starts without a stored token builds an initial token. Both versions default to starting at the beginning of the event stream, so a fresh processor handles the entire history in both. What changed is the ReplayStatus under which that history is handled.

Axon Framework 4 wrapped the initial token in a ReplayToken that concluded at the head of the stream, marking the whole catch-up as a replay:

// AF4, PooledStreamingEventProcessor.Builder default
initialToken = messageSource -> ReplayToken.createReplayToken(messageSource.createHeadToken());

Axon Framework 5 concludes that replay at the first position of the stream instead, which means it is already concluded by the first event handled:

// AF5, PooledStreamingEventProcessorConfiguration default
initialToken = eventSource -> eventSource.firstToken(null)
                                         .thenApply(ReplayToken::createReplayToken);

Axon Framework 4 Axon Framework 5

	Axon Framework 4	Axon Framework 5
Start position of a fresh processor	Beginning of the stream	Beginning of the stream
`ReplayStatus` while catching up	`REPLAY` until the head at initialization	`REGULAR` for every event
Handlers that opt out of replays	Skip the initial catch-up	Handle the initial catch-up
Replay-concluded notification	Published when the head at initialization is reached	Not published, as there is no replay to conclude

Start position of a fresh processor

Beginning of the stream

ReplayStatus while catching up

REPLAY until the head at initialization

REGULAR for every event

Handlers that opt out of replays

Skip the initial catch-up

Handle the initial catch-up

Replay-concluded notification

Published when the head at initialization is reached

Not published, as there is no replay to conclude

This is the expected behavior for a brand-new processor, not an oversight. A processor starting for the first time has no previously built state to correct, so its first pass over the stream is a first pass, not a repetition of one. Replay semantics exist to signal "you are seeing these events again", which does not apply here.

Restoring the Axon Framework 4 behavior

Configure the initial token explicitly if your handlers rely on ReplayStatus.REPLAY to suppress side effects while a fresh deployment catches up on an existing event store. Passing the latest position as the token to conclude at reproduces the Axon Framework 4 default:

import org.axonframework.messaging.core.configuration.MessagingConfigurer;
import org.axonframework.messaging.eventhandling.processing.streaming.token.ReplayToken;

public class ReplayingInitialTokenExample {

    public void configure(MessagingConfigurer configurer) {
        configurer.eventProcessing(
                eventProcessing -> eventProcessing.pooledStreaming(
                        pooledStreaming -> pooledStreaming.processor(
                                "my-processor",
                                module -> module.eventHandlingComponents(
                                                         components -> components.autodetected(
                                                                 "my-projector",
                                                                 cfg -> new MyProjector()
                                                         )
                                                 )
                                                 .customized((cfg, processorConfig) -> processorConfig.initialToken(
                                                         eventSource -> eventSource.latestToken(null)
                                                                                   .thenApply(ReplayToken::createReplayToken)
                                                 ))
                        )
                )
        );
    }
}

Note that the position to conclude at is resolved once, when the processor initializes its tokens, and is stored inside the token itself. Events appended after that moment are handled as regular, live events.

`QueryUpdateEmitter` as a parameter

In Axon Framework 4, the QueryUpdateEmitter was often injected as a class-level field in projectors. In Axon Framework 5, the use of QueryUpdateEmitter is enforced as a parameter of message handlers (specifically query handlers and event handlers that emit updates).

// AF4
public class MyProjector {
    @Autowired
    private QueryUpdateEmitter updateEmitter;

    @EventHandler
    public void on(MyEvent event) {
        updateEmitter.emit(MyQuery.class, q -> true, new MyView());
    }
}

// AF5
import org.axonframework.messaging.eventhandling.annotation.EventHandler;
import org.axonframework.messaging.queryhandling.QueryUpdateEmitter;

public class MyProjector {

    @EventHandler
    public void on(MyEvent event, QueryUpdateEmitter updateEmitter) {
        updateEmitter.emit(MyQuery.class, q -> true, new MyView());
    }
}

This change makes the dependency on the update emitter explicit for each handler and simplifies testing and component isolation.