Axonframework: How to synchronize command persisting.

Created on 25 Jun 2019 · 3Comments · Source: AxonFramework/AxonFramework

Hello,

I create test, that call over microservice by rest in 10 threats at same time (10 calls in one thread). I can see in logs command storage into DB throw exception -> same sequence_number. Have axon tool/pattern how to synchronize command persisting?

We are using:

org.axonframework
axon-spring-boot-starter
3.1.2

axon.eventhandling.processors.mygroup.mode = subscribing

In Service module, we call:
commandGateway.sendAndWait with command

aggregate is only string:
@AggregateIdentifier
private String uid;

2019-06-25T10:48:46,364Z [http-nio-8092-exec-10] ERROR o.a.c.c.C.[.[.[.[dispatcherServlet] - Servlet.service() for servlet [dispatcherServlet] in context with path [] threw exception [Request processing failed; nested exception is org.axonframework.commandhandling.model.ConcurrencyException: An event for aggregate [8fd8e2a1-3abf-4180-93c4-a719b9d71275] at sequence [5] was already inserted] with root cause
com.mysql.jdbc.exceptions.jdbc4.MySQLIntegrityConstraintViolationException: Duplicate entry '8fd8e2a1-3abf-4180-93c4-a719b9d71275-5' for key 'UK8s1f994p4la2ipb13me2xqm1w'
at sun.reflect.NativeConstructorAccessorImpl.newInstance0(Native Method) ~[na:1.8.0_144]
at sun.reflect.NativeConstructorAccessorImpl.newInstance(NativeConstructorAccessorImpl.java:62) ~[na:1.8.0_144]
at sun.reflect.DelegatingConstructorAccessorImpl.newInstance(DelegatingConstructorAccessorImpl.java:45) ~[na:1.8.0_144]
at java.lang.reflect.Constructor.newInstance(Constructor.java:423) ~[na:1.8.0_144]
at com.mysql.jdbc.Util.handleNewInstance(Util.java:425) ~[mysql-connector-java-5.1.44.jar:5.1.44]
at com.mysql.jdbc.Util.getInstance(Util.java:408) ~[mysql-connector-java-5.1.44.jar:5.1.44]
at com.mysql.jdbc.SQLError.createSQLException(SQLError.java:935) ~[mysql-connector-java-5.1.44.jar:5.1.44]
at com.mysql.jdbc.MysqlIO.checkErrorPacket(MysqlIO.java:3973) ~[mysql-connector-java-5.1.44.jar:5.1.44]
at com.mysql.jdbc.MysqlIO.checkErrorPacket(MysqlIO.java:3909) ~[mysql-connector-java-5.1.44.jar:5.1.44]
at com.mysql.jdbc.MysqlIO.sendCommand(MysqlIO.java:2527) ~[mysql-connector-java-5.1.44.jar:5.1.44]
at com.mysql.jdbc.MysqlIO.sqlQueryDirect(MysqlIO.java:2680) ~[mysql-connector-java-5.1.44.jar:5.1.44]
at com.mysql.jdbc.ConnectionImpl.execSQL(ConnectionImpl.java:2487) ~[mysql-connector-java-5.1.44.jar:5.1.44]
at com.mysql.jdbc.PreparedStatement.executeInternal(PreparedStatement.java:1858) ~[mysql-connector-java-5.1.44.jar:5.1.44]
at com.mysql.jdbc.PreparedStatement.executeUpdateInternal(PreparedStatement.java:2079) ~[mysql-connector-java-5.1.44.jar:5.1.44]
at com.mysql.jdbc.PreparedStatement.executeUpdateInternal(PreparedStatement.java:2013) ~[mysql-connector-java-5.1.44.jar:5.1.44]
at com.mysql.jdbc.PreparedStatement.executeLargeUpdate(PreparedStatement.java:5104) ~[mysql-connector-java-5.1.44.jar:5.1.44]
at com.mysql.jdbc.PreparedStatement.executeUpdate(PreparedStatement.java:1998) ~[mysql-connector-java-5.1.44.jar:5.1.44]
at sun.reflect.GeneratedMethodAccessor402.invoke(Unknown Source) ~[na:na]
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) ~[na:1.8.0_144]
at java.lang.reflect.Method.invoke(Method.java:498) ~[na:1.8.0_144]
at org.apache.tomcat.jdbc.pool.StatementFacade$StatementProxy.invoke(StatementFacade.java:114) ~[tomcat-jdbc-8.5.23.jar:na]
at com.sun.proxy.$Proxy255.executeUpdate(Unknown Source) ~[na:na]
at org.hibernate.engine.jdbc.internal.ResultSetReturnImpl.executeUpdate(ResultSetReturnImpl.java:204) ~[hibernate-core-5.0.12.Final.jar:5.0.12.Final]
at org.hibernate.dialect.identity.GetGeneratedKeysDelegate.executeAndExtract(GetGeneratedKeysDelegate.java:57) ~[hibernate-core-5.0.12.Final.jar:5.0.12.Final]
at org.hibernate.id.insert.AbstractReturningDelegate.performInsert(AbstractReturningDelegate.java:42) ~[hibernate-core-5.0.12.Final.jar:5.0.12.Final]
at org.hibernate.persister.entity.AbstractEntityPersister.insert(AbstractEntityPersister.java:2855) ~[hibernate-core-5.0.12.Final.jar:5.0.12.Final]
at org.hibernate.persister.entity.AbstractEntityPersister.insert(AbstractEntityPersister.java:3426) ~[hibernate-core-5.0.12.Final.jar:5.0.12.Final]
at org.hibernate.action.internal.EntityIdentityInsertAction.execute(EntityIdentityInsertAction.java:81) ~[hibernate-core-5.0.12.Final.jar:5.0.12.Final]
at org.hibernate.engine.spi.ActionQueue.execute(ActionQueue.java:619) ~[hibernate-core-5.0.12.Final.jar:5.0.12.Final]
at org.hibernate.engine.spi.ActionQueue.addResolvedEntityInsertAction(ActionQueue.java:273) ~[hibernate-core-5.0.12.Final.jar:5.0.12.Final]
at org.hibernate.engine.spi.ActionQueue.addInsertAction(ActionQueue.java:254) ~[hibernate-core-5.0.12.Final.jar:5.0.12.Final]
at org.hibernate.engine.spi.ActionQueue.addAction(ActionQueue.java:299) ~[hibernate-core-5.0.12.Final.jar:5.0.12.Final]
at org.hibernate.event.internal.AbstractSaveEventListener.addInsertAction(AbstractSaveEventListener.java:317) ~[hibernate-core-5.0.12.Final.jar:5.0.12.Final]
at org.hibernate.event.internal.AbstractSaveEventListener.performSaveOrReplicate(AbstractSaveEventListener.java:272) ~[hibernate-core-5.0.12.Final.jar:5.0.12.Final]
at org.hibernate.event.internal.AbstractSaveEventListener.performSave(AbstractSaveEventListener.java:178) ~[hibernate-core-5.0.12.Final.jar:5.0.12.Final]
at org.hibernate.event.internal.AbstractSaveEventListener.saveWithGeneratedId(AbstractSaveEventListener.java:109) ~[hibernate-core-5.0.12.Final.jar:5.0.12.Final]
at org.hibernate.jpa.event.internal.core.JpaPersistEventListener.saveWithGeneratedId(JpaPersistEventListener.java:67) ~[hibernate-entitymanager-5.0.12.Final.jar:5.0.12.Final]
at org.hibernate.event.internal.DefaultPersistEventListener.entityIsTransient(DefaultPersistEventListener.java:189) ~[hibernate-core-5.0.12.Final.jar:5.0.12.Final]
at org.hibernate.event.internal.DefaultPersistEventListener.onPersist(DefaultPersistEventListener.java:132) ~[hibernate-core-5.0.12.Final.jar:5.0.12.Final]
at org.hibernate.event.internal.DefaultPersistEventListener.onPersist(DefaultPersistEventListener.java:58) ~[hibernate-core-5.0.12.Final.jar:5.0.12.Final]
at org.hibernate.internal.SessionImpl.firePersist(SessionImpl.java:775) ~[hibernate-core-5.0.12.Final.jar:5.0.12.Final]
at org.hibernate.internal.SessionImpl.persist(SessionImpl.java:748) ~[hibernate-core-5.0.12.Final.jar:5.0.12.Final]
at org.hibernate.internal.SessionImpl.persist(SessionImpl.java:753) ~[hibernate-core-5.0.12.Final.jar:5.0.12.Final]
at org.hibernate.jpa.spi.AbstractEntityManagerImpl.persist(AbstractEntityManagerImpl.java:1146) ~[hibernate-entitymanager-5.0.12.Final.jar:5.0.12.Final]
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) ~[na:1.8.0_144]
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62) ~[na:1.8.0_144]
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) ~[na:1.8.0_144]
at java.lang.reflect.Method.invoke(Method.java:498) ~[na:1.8.0_144]
at org.springframework.orm.jpa.ExtendedEntityManagerCreator$ExtendedEntityManagerInvocationHandler.invoke(ExtendedEntityManagerCreator.java:347) ~[spring-orm-4.3.12.RELEASE.jar:4.3.12.RELEASE]
at com.sun.proxy.$Proxy196.persist(Unknown Source) ~[na:na]
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) ~[na:1.8.0_144]
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62) ~[na:1.8.0_144]
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) ~[na:1.8.0_144]
at java.lang.reflect.Method.invoke(Method.java:498) ~[na:1.8.0_144]
at org.springframework.orm.jpa.SharedEntityManagerCreator$SharedEntityManagerInvocationHandler.invoke(SharedEntityManagerCreator.java:298) ~[spring-orm-4.3.12.RELEASE.jar:4.3.12.RELEASE]
at com.sun.proxy.$Proxy196.persist(Unknown Source) ~[na:na]
at java.util.stream.ForEachOps$ForEachOp$OfRef.accept(ForEachOps.java:184) ~[na:1.8.0_144]
at java.util.stream.ReferencePipeline$3$1.accept(ReferencePipeline.java:193) ~[na:1.8.0_144]
at java.util.ArrayList$ArrayListSpliterator.forEachRemaining(ArrayList.java:1374) ~[na:1.8.0_144]
at java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:481) ~[na:1.8.0_144]
at java.util.stream.AbstractPipeline.wrapAndCopyInto(AbstractPipeline.java:471) ~[na:1.8.0_144]
at java.util.stream.ForEachOps$ForEachOp.evaluateSequential(ForEachOps.java:151) ~[na:1.8.0_144]
at java.util.stream.ForEachOps$ForEachOp$OfRef.evaluateSequential(ForEachOps.java:174) ~[na:1.8.0_144]
at java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:234) ~[na:1.8.0_144]
at java.util.stream.ReferencePipeline.forEach(ReferencePipeline.java:418) ~[na:1.8.0_144]
at org.axonframework.eventsourcing.eventstore.jpa.JpaEventStorageEngine.appendEvents(JpaEventStorageEngine.java:362) ~[axon-core-3.3.jar:3.3]
at org.axonframework.eventsourcing.eventstore.AbstractEventStorageEngine.appendEvents(AbstractEventStorageEngine.java:119) ~[axon-core-3.3.jar:3.3]
at org.axonframework.eventsourcing.eventstore.AbstractEventStore.prepareCommit(AbstractEventStore.java:66) ~[axon-core-3.3.jar:3.3]
at org.axonframework.eventhandling.AbstractEventBus.doWithEvents(AbstractEventBus.java:220) ~[axon-core-3.3.jar:3.3]
at org.axonframework.eventhandling.AbstractEventBus.lambda$null$8(AbstractEventBus.java:155) ~[axon-core-3.3.jar:3.3]
at org.axonframework.messaging.unitofwork.MessageProcessingContext.notifyHandlers(MessageProcessingContext.java:68) ~[axon-core-3.3.jar:3.3]
at org.axonframework.messaging.unitofwork.DefaultUnitOfWork.notifyHandlers(DefaultUnitOfWork.java:91) ~[axon-core-3.3.jar:3.3]
at org.axonframework.messaging.unitofwork.AbstractUnitOfWork.changePhase(AbstractUnitOfWork.java:221) ~[axon-core-3.3.jar:3.3]
at org.axonframework.messaging.unitofwork.AbstractUnitOfWork.commitAsRoot(AbstractUnitOfWork.java:82) ~[axon-core-3.3.jar:3.3]
at org.axonframework.messaging.unitofwork.AbstractUnitOfWork.commit(AbstractUnitOfWork.java:70) ~[axon-core-3.3.jar:3.3]
at org.axonframework.messaging.unitofwork.DefaultUnitOfWork.executeWithResult(DefaultUnitOfWork.java:80) ~[axon-core-3.3.jar:3.3]
at org.axonframework.commandhandling.SimpleCommandBus.handle(SimpleCommandBus.java:148) ~[axon-core-3.3.jar:3.3]
at org.axonframework.commandhandling.SimpleCommandBus.doDispatch(SimpleCommandBus.java:121) ~[axon-core-3.3.jar:3.3]
at org.axonframework.commandhandling.SimpleCommandBus.dispatch(SimpleCommandBus.java:85) ~[axon-core-3.3.jar:3.3]
at org.axonframework.commandhandling.gateway.AbstractCommandGateway.send(AbstractCommandGateway.java:79) ~[axon-core-3.3.jar:3.3]
at org.axonframework.commandhandling.gateway.DefaultCommandGateway.send(DefaultCommandGateway.java:95) ~[axon-core-3.3.jar:3.3]
at org.axonframework.commandhandling.gateway.DefaultCommandGateway.sendAndWait(DefaultCommandGateway.java:113) ~[axon-core-3.3.jar:3.3]

Question

Source

Srsnik

All 3 comments

Hi Peter,

Firstly, I want to stress that framework usage questions like this should be targeted towards StackOverflow with the axon tag or on the User Group. If more direct support is required, you could contact AxonIQ as well of course, but I assume that'll no be necessary.

Regardless, I'll try to help you out with the question you're having, after which I _will_ be closing this issue as a question.

From the shared stack trace and scenario, I am assuming the test case dispatches 10 commands for the same aggregate, correct?
The framework will internally load an Aggregate upon receiving a command for it, which in most scenarios, like yours, results in publishing an event.
As the events should be published in order, the framework puts a unique key constraint on the aggregateId-sequenceNumber pair.

As such, if you'd have several services trying to access the same Aggregate, you would see this Constraint Violation Exception, which is marked as a ConcurrencyException.
It does so, as there is wrongful concurrent communication on an Aggregate, which is undesirable.

To resolve this, it is recommended to either use Axon Server or set up a DistributedCommandBus (the latter through either [JGroups]https://github.com/AxonFramework/extension-jgroups] or Spring Cloud. Axon Server is a purpose build Routing mechanism for both Commands, Events and Queries, as well as a dedicated Event Store, with monitoring provided on top of this.

Both approach _are free_ to use, but more importantly both approaches will ensure that a command will always be routed to the same instance.
This will thus ensure that it's the same thread/application/microservice that will be in charge of all 10 commands, regardless which thread/application/microservice dispatched the command.

Hope this helps you out @Srsnik!