We are using Axon 4 and have a multi-module project structure with separated database schemas. We have an axon module which provides a global command gateway (and a central event store). Each module in our project acts as an Axon client.
Now we have various aggregates in each module which also act as command handlers. A command can for instance create a new aggregate in a module. As long as the command comes from a class from within the module everything works fine.
However, now one module sends a command via the global command bus and triggers the creation of a new aggregate in another module. Now Axon logs a warning that no transaction is in progress. This seems logical as the class which sends the command is managed by another transaction manager. But how do we solve this? A constructor cannot be marked with @Transactional. Do we really need to add another command handler between those classes to make sure that we have a transaction? How would this even work in a distributed environment?
by default Axon will configure “the” transaction manager on the command bus, to ensure a transaction is started. It looks like you have multiple instances. In that case, you need to make sure that the correct instance is being used.
It sounds like you also have multiple instances of the CommandBus in your context. Is that correct? How did you configure them?
Note that in a distributed setup, you wouldn’t have this problem, as each instance will have its own transaction manager, making it out of the box.
Currently we use only a single command bus. Our project is actually a Spring Boot application with hierarchical application contexts. One of those contexts is an Axon embedded server module which exports the command gateway (and thus the command bus) to the parent context. The other client modules (which are actually siblings and not children of the Axon module) use this exported command gateway. As each of those client modules bring their own transaction manager it is probably not possible to replace the transaction manager of the UnitOfWork, right?
We assume by now that we should probably have a command bus for each client module and some kind of “parent” command bus which would distribute the commands to the individual command busses. However, we are not sure how to do this. We know that there is a distributed command bus, but can this be used in a non-distributed environment (in other words: in the same JVM) and with the embedded Axon server?
it shouldn’t be necessary to create a command bus per module. Instead, the transaction manager on that command bus must be compatible with either of the two databases used by the different modules.
Not sure how that would work on the Spring-level, though. Axon takes “the transaction manager” from the application context when configuring the command bus. It seems to be able to find it, but that transaction manager is not compatible with the datasource used in some of the modules.
I don’t think using the DistributedCommandBus is a solution in this case. It will most likely only complicate things more.
There are two modules (module1 and module2). Each of them represents a bounded context. module2 has a state-stored aggregate called Module2Aggregate. The creation of this aggregate is triggered by the CreateModule2AggregateCommand which is sent by the Module1CommandSender located in module1.
Both, module1 and module2, have a seperate Spring application context which are children of the root context (see AxonHierarchicalApplication class). The third child module (called axon) currently provided the infrasturcture part (ComandBus, CommandGateway, EventBus,…). These beans are exported (by the ExportedBeanPostProcessor) to the root context and can therefore be used from the other child contexts.
Notice: Each module (axon, module1, module2) does have a seperate datasource and a seperate JPA transaction manager. They do it this way in order to decouple the persistence of the modules as much as possible. This enables the possibility to decide at any point of time to run those modules in one application (single deployment unit) or in a distributed way.
What happens when you run the sample application:
de.libutzki.axon.axonhierarchical.module1.Module1CommandSender.sendCommand() is invoked by de.libutzki.axon.axonhierarchical.AxonHierarchicalApplication.main(String)
Because of the @Transactional annotation a new transaction is started by the module1 transaction manager.
Module1CommandSender sends the CreateModule2AggregateCommand via the CommandGateway which is provided (exported to the root context) by the axon module.
The CommandGateway delegates the command to the SimpleCommandBus which is initialized the the transaction manager of the axon module.
A transaction is started by the axon transaction manager
The CommandHandler de.libutzki.axon.axonhierarchical.module2.aggregate.Module2Aggregate.Module2Aggregate(CreateModule2AggregateCommand) is invoked,
The method invocation works, but after that an Exception is thrown: javax.persistence.TransactionRequiredException: no transaction is in progress
(Full stacktrace attached at the end)
It’s obvious why this exception is thrown: The axon transaction manager started the transaction, but we expect the module2 transaction manager to start the transaction.
As the CommandBus is bound to a single transaction manager we think about having a command bus in each child module. Every command bus is bound to the child module’s transaction manager.
When having multiple command buses, we need to introduce a central command bus which orchestrates all the child command buses concerning handler subscriptions and message dispatching. We think it’s similar to what a DistributedCommandBus does. Anyway, it might not be the right choice to reuse the DistributedCommandBus as concepts like the load factor are not relevant for us. So having a new command bus implementation for orchestrating our independent child command buses might be the best solution.
I hope you got our point. If you have any suggestion how to solve this, we are very keen to know.
Thanks for your support!
Kind regards
Oliver
Stacktrace:
`
Exception in thread "main" javax.persistence.TransactionRequiredException: no transaction is in progress
at org.hibernate.internal.AbstractSharedSessionContract.checkTransactionNeededForUpdateOperation(AbstractSharedSessionContract.java:398)
at org.hibernate.internal.SessionImpl.checkTransactionNeededForUpdateOperation(SessionImpl.java:3550)
at org.hibernate.internal.SessionImpl.doFlush(SessionImpl.java:1443)
at org.hibernate.internal.SessionImpl.flush(SessionImpl.java:1439)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:498)
at org.springframework.orm.jpa.ExtendedEntityManagerCreator$ExtendedEntityManagerInvocationHandler.invoke(ExtendedEntityManagerCreator.java:350)
at com.sun.proxy.$Proxy70.flush(Unknown Source)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:498)
at org.springframework.orm.jpa.SharedEntityManagerCreator$SharedEntityManagerInvocationHandler.invoke(SharedEntityManagerCreator.java:308)
at com.sun.proxy.$Proxy70.flush(Unknown Source)
at org.axonframework.modelling.command.GenericJpaRepository.doSaveWithLock(GenericJpaRepository.java:145)
at org.axonframework.modelling.command.GenericJpaRepository.doSaveWithLock(GenericJpaRepository.java:53)
at org.axonframework.modelling.command.LockingRepository.doSave(LockingRepository.java:149)
at org.axonframework.modelling.command.LockingRepository.doSave(LockingRepository.java:52)
at org.axonframework.modelling.command.AbstractRepository.doCommit(AbstractRepository.java:194)
at org.axonframework.modelling.command.AbstractRepository.prepareForCommit(AbstractRepository.java:180)
at org.axonframework.modelling.command.LockingRepository.prepareForCommit(LockingRepository.java:131)
at org.axonframework.modelling.command.LockingRepository.prepareForCommit(LockingRepository.java:52)
at org.axonframework.modelling.command.AbstractRepository.lambda$newInstance$0(AbstractRepository.java:86)
at org.axonframework.messaging.unitofwork.MessageProcessingContext.notifyHandlers(MessageProcessingContext.java:71)
at org.axonframework.messaging.unitofwork.DefaultUnitOfWork.notifyHandlers(DefaultUnitOfWork.java:106)
at org.axonframework.messaging.unitofwork.AbstractUnitOfWork.changePhase(AbstractUnitOfWork.java:222)
at org.axonframework.messaging.unitofwork.AbstractUnitOfWork.commitAsRoot(AbstractUnitOfWork.java:83)
at org.axonframework.messaging.unitofwork.AbstractUnitOfWork.commit(AbstractUnitOfWork.java:71)
at org.axonframework.messaging.unitofwork.DefaultUnitOfWork.executeWithResult(DefaultUnitOfWork.java:92)
at org.axonframework.commandhandling.SimpleCommandBus.handle(SimpleCommandBus.java:176)
at org.axonframework.commandhandling.SimpleCommandBus.doDispatch(SimpleCommandBus.java:141)
at org.axonframework.commandhandling.SimpleCommandBus.dispatch(SimpleCommandBus.java:110)
at org.axonframework.commandhandling.gateway.AbstractCommandGateway.send(AbstractCommandGateway.java:75)
at org.axonframework.commandhandling.gateway.DefaultCommandGateway.send(DefaultCommandGateway.java:73)
at org.axonframework.commandhandling.gateway.DefaultCommandGateway.sendAndWait(DefaultCommandGateway.java:90)
at de.libutzki.axon.axonhierarchical.module1.Module1CommandSender.sendCommand(Module1CommandSender.java:27)
at de.libutzki.axon.axonhierarchical.module1.Module1CommandSender$$FastClassBySpringCGLIB$$287801a8.invoke(<generated>)
at org.springframework.cglib.proxy.MethodProxy.invoke(MethodProxy.java:218)
at org.springframework.aop.framework.CglibAopProxy$CglibMethodInvocation.invokeJoinpoint(CglibAopProxy.java:749)
at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:163)
at org.springframework.transaction.interceptor.TransactionAspectSupport.invokeWithinTransaction(TransactionAspectSupport.java:294)
at org.springframework.transaction.interceptor.TransactionInterceptor.invoke(TransactionInterceptor.java:98)
at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:186)
at org.springframework.aop.framework.CglibAopProxy$DynamicAdvisedInterceptor.intercept(CglibAopProxy.java:688)
at de.libutzki.axon.axonhierarchical.module1.Module1CommandSender$$EnhancerBySpringCGLIB$$683db6d8.sendCommand(<generated>)
at de.libutzki.axon.axonhierarchical.AxonHierarchicalApplication.main(AxonHierarchicalApplication.java:24)
I finally fixed this issue by introducing a DelegatingCommandBus. This command bus is provided by the axon module. module1 and module2 own seperate command buses. These command buses register at the DelegatingCommandBus. I had to extend the CommandBus interface in order to ask the command bus if it could handle a certain CommandMessage. This job is done by the DelegatingCommandBus. It askes every registered command bus if it could handle the command and delegates the command afterwards.
I highly appreciate if you could integrate this (or a similar) solution in Axon. The focus of my implementation might not address the issue entirely as it was created from a concrete project’s point of view. Anyway, I tried to abstract it in my sample GitHub project. You can find the solution in the delegating-command-bus branch.
I’ve been spending some thought-cycles on this one as well. I came to a slightly different approach, which should avoid the need for a “canHandle” type operation. My main reason to avoid it is that it is hard to combine with a distributed setup.
What I came up with, is a “FederatedCommandBus” implementation (perhaps HierarchicalCommandBus could be an alternative name). It allows the registration of a CommandBus instance that does the actual execution of the handlers (which would default to a SimpleCommandBus) as well as (optionally) a parent CommandBus. Strictly speaking, the parent doesn’t need to be a FederatedCommandBus.
When you subscribe a handler on a FederatedCommandBus instance, it will register the handler with its local “CommandBus”. If the FederatedCommandBus has a parent configured, it will also register itself as the handler of the command with the parent, so that the parent forwards commands of that type to that instance.
When dispatching a command to a FederatedCommandBus, you could take 2 approaches. Either check locally first if there is a handler. If not, delegate to the parent instance. The other solution is to always delegate to the parent, and let that delegate it back if necessary.
This is a mechanism that I believe would benefit the framework in general, so I would be happy to include this in the framework.
after thinking about your idea again, I came to the conclusion that it doesn’t work for our use case.
All the commands which are sent to the parent command bus are dispatched by this command bus and therefore this results in the problem that the TransactionManager of the parent command bus is used.
This results in a “no transaction is in progress” error again. So, it’s mandatory that the CommandBus to which a certain Handler subscribes dispatches the command. In this case the correct TransactionManager is attached to the UnitOfWork.
the idea isn’t that the parent bus invokes any handlers. A child bus should register itself as the handler for commands with the parent bus. In that case, it can force all message handler interceptors (incl the one that starts the transaction) to be invoked.
Ok, I have to admit I don't understand how you can achieve that, but I'm keen to know. For the moment my solution works for us, but of course I'd like to switch to a concept Axon supports natively.
