JpaTokenStore causes event handler not to receive events

Imported from Google Groups
#1

I am trying to configure a tracking event processor to keep track of events that were processed when the application restarts.

This is my configuration
`
Configurer configurer = DefaultConfigurer.defaultConfiguration()
.configureAggregate(MyAggregate.class)
.eventProcessing(eventProcessingConfigurer -> {

eventProcessingConfigurer.registerTokenStore(“default”, configuration -> JpaTokenStore.builder().entityManagerProvider(new SimpleEntityManagerProvider(emf.createEntityManager())).serializer(configuration.serializer()).build());
// If I comment out the above line the event handler works as expected. I assume it defaults to the InMemoryTokenStore which is why the events are processed every time the application restarts.

eventProcessingConfigurer.registerTrackingEventProcessor(“default”);
eventProcessingConfigurer.registerEventHandler(configuration -> new AggregateEventHandler());
});
Configuration configuration = configurer.start();
`

This is my event handler
`
@ProcessingGroup(“default”)
public class AggregateEventHandler {

@EventHandler
public void on(AggregateRegisteredEvent event) {
System.out.println(“Registered Event”);
}

@EventHandler
public void on(AggregateUpdatedEvent event) {
System.out.println(“Updated Event”);
}

}

`

From the log I can see that the JPA connection is successful and the TOKENENTRY table is created but no record is inserted.
Am I missing a configuration?

Thanks

#2

Hi Cobus,

generally, JPA will need some form of transaction to know when it should commit changes made in the entity manager. My suspisicion is that, since you’re using the SimpleEntityManagerProvider, these changes are never flushed to the database.

We don’t have explicit implementation for application managed EntityManagers yet, as 99.9% of our userbase uses some sort of CDI container. Basiically, your EntityManagerProvider will need to create a new EnittyManager, and make sure it is (flushed and) closed at a proper time. You can leverage the Unit Of Work to do that. You can register an onCommit() handler to flush and close and the onRollback() to roll back any changes.

If you’re using a Container Managed EntityManager, you can generally just use the SimpleEntityManagerProvider, as there is a single instance for the entire container.

Hope this helps.
Cheers,

#3

Hi Allard

I am trying to use Axon inside an Apache Karaf OSGI container so I should be able to inject an entity manager.
However I am receiving the following errors:

2019-01-31 09:48:56,236 | WARN | .domain.query]-0 | TrackingEventProcessor | 135 - org.axonframework.axon-messaging - 4.0.2 | Fetch Segments for Processor ‘test.axon.domain.query’ failed: Need active coordination. Preparing for retry in 1s
2019-01-31 09:48:56,251 | WARN | rint Extender: 1 | DefaultCommandGateway | 135 - org.axonframework.axon-messaging - 4.0.2 | Command ‘test.axon.domain.command.RegisterAggregateCommand’ resulted in org.axonframework.commandhandling.NoHandlerForCommandException(No handler was subscribed to command [test.axon.domain.command.RegisterAggregateCommand])

I noticed that all the axon dependencies are OSGI compliant except the axon-server-connector. Therefore I have to configure the event store to axon server manually because it was giving this error:

IllegalStateException: Default configuration requires the use of event sourcing. Either configure an Event Store to use, or configure a specific repository implementation for class test.axon.domain.MyAggregate

Any idea why I am receiving these errors?

Kind Regards

#4

I see a few issues here. First, the complaint about the “Default configuration requires the use of event sourcing”, is because unless you explicitly provide a repository for your aggregate, Axon assumes you will attempt to use an EventStore. It seems that there is no Event Store configured.

The NoHandlerForCommand indicates that no handler is registered with the Command Bus that can take that command. Did you manually configure AxonServer?

Last, there is the “Need active coordination” error, which is related to Karaf (see https://stackoverflow.com/questions/38327381/apache-karaf-cant-inject-entity-manager). It seems you need to explicitly start a transaction, similar to how the SpringTransactionManager does that with Spring. Register that TransactionManager with the configuration, and you should be all set.

Hope this helps.
Cheers,

Allard

#5

I have managed to fix the active coordination issue and my JpaEventStore is no longer giving any errors. However the NoHandlerForCommand issue is still there. I am not configuring Axon Server manually.
This is my configuration:

`
final EntityManagerProvider entityManagerProvider = new EntityManagerProvider() {
public EntityManager getEntityManager() {
return jpaEntityManagerProvider.getResource(transactionControl);
}
};

Configurer configurer = DefaultConfigurer.defaultConfiguration()
.registerComponent(EntityManagerProvider.class, new Function<Configuration, EntityManagerProvider>() {
public EntityManagerProvider apply(Configuration configuration) {
return entityManagerProvider;
}
})
.registerComponent(TransactionManager.class, new Function<Configuration, TransactionManager>() {
public TransactionManager apply(Configuration configuration) {
return new ContainerTransactionManager(transactionControl);
}
})
.configureEmbeddedEventStore(new Function<Configuration, EventStorageEngine>() {
public EventStorageEngine apply(Configuration configuration) {
return new TxAwareJPAEventStorageEngine(JpaEventStorageEngine.builder()
.snapshotSerializer(configuration.serializer())
.upcasterChain(configuration.upcasterChain())
.persistenceExceptionResolver(configuration.getComponent(PersistenceExceptionResolver.class))
.eventSerializer(configuration.eventSerializer())
.entityManagerProvider(configuration.getComponent(EntityManagerProvider.class))
.transactionManager(configuration.getComponent(TransactionManager.class))
);
}
})
.registerComponent(TokenStore.class, new Function<Configuration, TokenStore>() {
public TokenStore apply(Configuration configuration) {
return new TxAwareJpaTokenStore(JpaTokenStore.builder()
.entityManagerProvider(configuration.getComponent(EntityManagerProvider.class))
.serializer(configuration.serializer())
, transactionControl
);
}
})
.configureAggregate(MyAggregate.class)
.eventProcessing(new Consumer() {
public void accept(EventProcessingConfigurer configurer) {
configurer.registerEventHandler(new Function<Configuration, Object>() {
public Object apply(Configuration configuration) {
return new MyEventHandler();
}
});
}
})
;
Configuration configuration = configurer.start();
`

Even configuring a command handler manually does not make a difference.
If I run it outside of Karaf it is working so I assume it must be class loader related.

Any suggestions?

Thanks
Cobus

#6

Hi Cobus,

From the configuration snippet you’ve shared, I see that you’re not yet configuring the command bus explicitly.
As you’ve noted, the axon-server-connector module is not OSGi compliant, so I am assuming it’s something in that sphere which causes your Aggregate Command Handlers to not be registered to the right CommandBus.
I’d thus suggest to also introduce the configureCommandBus(Function<Configuration, CommandBus>) function to your `DefaultConfigurer.defaultConfiguration()

method call, prior to theconfigureAggregate(MyAggregate.class)` method call, to see if that will register your command handlers.

What you can also do to ensure the registration of your command handlers occurs as expected, is to debug this process.
The function being called to subscribe command handling functions to the command bus, is the CommandBus#subscribe(String commandName, MessageHandler<? super CommandMessage<?>> handler) method. Placing a break point in the function of the used implementation of the CommandBus, the AxonServerCommandBus, should be called with the command handling functions you’ve defined in the MyAggregate class.

I hope this helps you proceed Cobus!

Cheers,
Steven