How to enable Parallel Processing for a processing group using the Axon API?

1
votes

I'm trying to enable parallel processing on one of my processing group (a Saga to be exact), I followed the the Axon Reference and I ended up with the following

@Autowired
public void configureProcessor(Configurer configurer) {
    configurer.eventProcessing().registerTrackingEventProcessor("NameOfMySagaProcessor",
            org.axonframework.config.Configuration::eventStore,
            c -> c.getComponent(
                    TrackingEventProcessorConfiguration.class,
                    () -> TrackingEventProcessorConfiguration.forParallelProcessing(1).andInitialSegmentsCount(2).andInitialTrackingToken(StreamableMessageSource::createHeadToken)
            )
    );
}

I deleted the entry for this saga in my entry_token table so it repopulates everything but since the initial tracking token is a head Token then I'm not expecting the saga to replay all the events to reach head.

PS: This is the main reason why I didn't use the spring boot configuration since using the following doesn't allow you to select the initial tracking token

axon.eventhandling.processors.name.mode=tracking

axon.eventhandling.processors.name.threadCount=2

axon.eventhandling.processors.name.initialSegmentCount=4

The spring boot configuration worked but again without an initial tracking token whereas the api configuration didn't work as in nothing changed (my events weren't split into 4 segments...)

I am using the following version of axon

compile (group: 'org.axonframework', name: 'axon-spring-boot-starter', version: '4.0.3'){
    exclude group:'org.axonframework', module: 'axon-server-connector'
}
1
spring-bootdomain-driven-designevent-sourcingaxon

1 Answers

1
votes

Ended up fixing it by calling the following instead 

@Autowired
public void configureProcessor(Configurer configurer) {
    configurer.eventProcessing().registerTrackingEventProcessor("NameOfMySagaProcessor",
            org.axonframework.config.Configuration::eventStore, configuration -> TrackingEventProcessorConfiguration.forParallelProcessing(2).andInitialTrackingToken(StreamableMessageSource::createHeadToken)
    );
}