Axon Event Propagation

Imported from Google Groups
1

Hi. Guys.
I’m trying to event redelivery in Axon nowdays.

but I got some persist exception event if I just redelivery my event.

actually in my process,
when i got event, I will send Kafka message.
and in Kafka consumer, they will reformat my object to “EventMessage” object.
and delivery my event, and my event handler can get my second event.

but the problem is when i re-send throught command gate way in my event handler.
in my seconde event handler, just will resend my event for command gate way.

my error log is

org.axonframework.eventsourcing.eventstore.EventStoreException: An event for aggregate [A145SA0149] at sequence [1] could not be persisted
at org.axonframework.eventsourcing.eventstore.AbstractEventStorageEngine.handlePersistenceException(AbstractEventStorageEngine.java:112)
at org.axonframework.eventsourcing.eventstore.jpa.JpaEventStorageEngine.appendEvents(JpaEventStorageEngine.java:271)
at org.axonframework.eventsourcing.eventstore.AbstractEventStorageEngine.appendEvents(AbstractEventStorageEngine.java:85)
at org.axonframework.eventsourcing.eventstore.AbstractEventStore.prepareCommit(AbstractEventStore.java:64)
at org.axonframework.eventhandling.AbstractEventBus.doWithEvents(AbstractEventBus.java:210)
at org.axonframework.eventhandling.AbstractEventBus.lambda$null$4(AbstractEventBus.java:145)
at org.axonframework.messaging.unitofwork.MessageProcessingContext.notifyHandlers(MessageProcessingContext.java:68)
at org.axonframework.messaging.unitofwork.BatchingUnitOfWork.lambda$notifyHandlers$2(BatchingUnitOfWork.java:131)
at java.util.ArrayList$Itr.forEachRemaining(ArrayList.java:891)
at org.axonframework.messaging.unitofwork.BatchingUnitOfWork.notifyHandlers(BatchingUnitOfWork.java:131)

Caused by: javax.persistence.TransactionRequiredException: No EntityManager with actual transaction available for current thread - cannot reliably process ‘persist’ call
at org.springframework.orm.jpa.SharedEntityManagerCreator$SharedEntityManagerInvocationHandler.invoke(SharedEntityManagerCreator.java:282)
at com.sun.proxy.$Proxy152.persist(Unknown Source)
at java.util.stream.ForEachOps$ForEachOp$OfRef.accept(ForEachOps.java:184)
at java.util.stream.ReferencePipeline$3$1.accept(ReferencePipeline.java:193)
at java.util.ArrayList$ArrayListSpliterator.forEachRemaining(ArrayList.java:1374)
at java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:481)
at java.util.stream.AbstractPipeline.wrapAndCopyInto(AbstractPipeline.java:471)
at java.util.stream.ForEachOps$ForEachOp.evaluateSequential(ForEachOps.java:151)
at java.util.stream.ForEachOps$ForEachOp$OfRef.evaluateSequential(ForEachOps.java:174)
at java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:234)
at java.util.stream.ReferencePipeline.forEach(ReferencePipeline.java:418)
at org.axonframework.eventsourcing.eventstore.jpa.JpaEventStorageEngine.appendEvents(JpaEventStorageEngine.java:266)
… 36 more

Someone have some idea for sloving this problem.?

2

Hi Suij,

I’m not a 100% sure if I follow the issue you’re describing, but do think I know of a solution for the exception you’re seeing.
Have you tried registering a TransactionManagingInterceptor handler to your CommandBus?
In a config file where you define your CommandBus, that would look as follows:

@Bean
public CommandBus commandBus(TransactionManager transactionManager) {
CommandBus commandBus = new SimpleCommandBus(); // This can obviously be another command bus implementation as well
commandBus.registerHandlerInterceptor(new TransactionManagingInterceptor(transactionManager));
return commandBus;
}

That might just solve the TransactionRequiredException you’re seeing.

Hope this helps!

Cheers,

Steven