Concurrent update in mongo

Aymen_Kanzari · June 25, 2024, 5:47pm

In my aggregate i have the below commandHandler, in which i dispatch two events CatalogAdded and NoteAdded

    @CommandHandler
    private void handle(AddCatalog cmd) {
        AggregateLifecycle.apply(CatalogAdded.builder().catalogId(cmd.catalogId()).name(cmd.name()).build())
                .andThenApply(() -> NoteAdded.builder().catalogId(cmd.catalogId()).description(cmd.description()).build());
    }

i the below processing groups, in both event handler i get the product from mongo and i update it

@Service
@RequiredArgsConstructor
@ProcessingGroup(CATALOG_INIT_EVENT)
public class CatalogEventHandler {

    private final ProductRepository repository;

    @EventHandler
    public void on(CatalogAdded evt) {
	    repository.findByCatalogId(evt.catalogId()).ifPresent(catalog -> repository.save(catalog.toBuilder().name(evt.name()).build()));
	}
}

@Service
@RequiredArgsConstructor
@ProcessingGroup(CATALOG_DETAILS_EVENT)
public class CatalogDetailsEventHandler {

    private final ProductRepository repository;

    @EventHandler
    public void on(NoteAdded evt) {
	    repository.findByCatalogId(evt.catalogId()).ifPresent(catalog -> repository.save(catalog.toBuilder().description(evt.description()).build()));
	}
}

the problem is that the update in the database happens at the same time from the event handlers, for that i have this error

com.mongodb.MongoCommandException: Command failed with error 251 (NoSuchTransaction): ‘Transaction with { txnNumber: 182 } has been aborted.’ on server mongo-release-local:27017. The full response is {“errorLabels”: [“TransientTransactionError”], “ok”: 0.0, “errmsg”: “Transaction with { txnNumber: 182 } has been aborted.”, “code”: 251, “codeName”: “NoSuchTransaction”, “$clusterTime”: {“clusterTime”: {“$timestamp”: {“t”: 1719335834, “i”: 12}}, “signature”: {“hash”: {“$binary”: {“base64”: “Eb9sme94PgMw2YGhYm81RpUE6Yg=”, “subType”: “00”}}, “keyId”: 7358959517146546182}}, “operationTime”: {“$timestamp”: {“t”: 1719335834, “i”: 12}}}

Gerard · June 25, 2024, 6:20pm

I’m not sure why this would fail. Do you use a batch size for the event processor, can you set it to 1? Does that ‘fix’ the problem? Transactions in Mongo work less straightforward then with SQL, being added later on. Which database do you use, and do you use the spring boot starter from the Mongo extension?

Aymen_Kanzari · June 25, 2024, 7:44pm

I used spring boot, axon server and mongodb as database, and yes i use axon-mongo-spring-boot-starter
The two events CatalogAdded and NoteAdded are not in the same processing group (i updated my post), does this cause a problem ?
Can SequentialPolicy fix the problem?
How can i set the batch size and how can it fix the problem ?

event processor config:

final TrackingEventProcessorConfiguration tepConfig = TrackingEventProcessorConfiguration
                .forSingleThreadedProcessing().andInitialTrackingToken(StreamableMessageSource::createHeadToken);
configurer.registerTrackingEventProcessor(CATALOG_INIT_EVENT,  it -> MultiStreamableMessageSource.builder()
          .addMessageSource(myContext, (axonservereventstore).createStreamableMessageSourceForContext(dsContext)).build())
          .registerTrackingEventProcessorConfiguration(CATALOG_INIT_EVENT, config -> tepConfig);

Gerard · June 25, 2024, 8:42pm

If they are not in the same processing group, it makes a lot more sense. I’m that case both changes happen about the same time. I would prefer to keep any handlers updating/creating the same entities/documents in the same processing group. This way you should also be able to properly reset the handler, by throwing all the documents away, en read all the events again.

Aymen_Kanzari · June 25, 2024, 9:41pm

so, what do you propose in the below situation

   @CommandHandler
    private void handle(AddCatalog cmd) {
        AggregateLifecycle.apply(CatalogAdded.builder().catalogId(cmd.catalogId()).name(cmd.name()).build())
                .andThenApply(() -> NoteAdded.builder().catalogId(cmd.catalogId()).description(cmd.description()).build());
    }

   @CommandHandler
    private void handle(AddSku cmd) {
        AggregateLifecycle.apply(SkuAdded.builder().catalogId(cmd.catalogId()).skuId(cmd.skuId()).name(cmd.name()).build())
                .andThenApply(() -> NoteAdded.builder().catalogId(cmd.catalogId()).description(cmd.description()).build());
    }

i have a two different processing groups, one for catalog and the other for sku, so when i duplicate the event handler of NoteAdded in the two processing groups, when the NoteAdded event is dispatched it is treated 2 times, one time in the processing group of catalog and the other time in the processing group of sku, for this i put it in a separated processing group to be treated only once

Gerard · June 26, 2024, 2:53am

No, what I meant is that the event handlers would be in the same group. Whether in the same class or not, doesn’t really makes a difference.

For the command handling, I don’t really know, and it’s also not related to the problem.

rsobies · December 13, 2025, 7:20pm

I have similar problem.
I get the error message, like this

rg.axonframework.eventsourcing.eventstore.EventStoreException: An event for aggregate [7c302cbf-27d4-466d-a1d9-20bc8d5e2e16] at sequence [1] was already inserted

...

Caused by: com.mongodb.MongoCommandException: Command failed with error 251 (NoSuchTransaction): 'Transaction with { txnNumber: 17942 } has been aborted.' on server production1-shard-00-04-pri.7udvj.mongodb.net:27017. The full response is {"errorLabels": ["TransientTransactionError"], "ok": 0.0, "errmsg": "Transaction with { txnNumber: 17942 } has been aborted.", "code": 251, "codeName": "NoSuchTransaction", "$clusterTime": {"clusterTime": {"$timestamp": {"t": 1765650977, "i": 1468}}, "signature": {"hash": {"$binary": {"base64": "sKPAK8EPwg9865T6w5rbulpqdwE=", "subType": "00"}}, "keyId": 7534769034131145126}}, "operationTime": {"$timestamp": {"t": 1765650977, "i": 1467}}}

when I retry the same command, then the event is applied successfully.
why the transaction is aborted?
this error is hard to reproduce.

rsobies · December 15, 2025, 1:19pm

sometimes this error apears during deadlinehandler excution

org.axonframework.deadline.DeadlineException: Failed to process
	at org.axonframework.deadline.jobrunr.JobRunrDeadlineManager.execute(JobRunrDeadlineManager.java:252)
	at java.base/jdk.internal.reflect.DirectMethodHandleAccessor.invoke(Unknown Source)
	at java.base/java.lang.reflect.Method.invoke(Unknown Source)
	at org.jobrunr.server.runner.AbstractBackgroundJobRunner$BackgroundJobWorker.invokeJobMethod(AbstractBackgroundJobRunner.java:64)
	at org.jobrunr.server.runner.AbstractBackgroundJobRunner$BackgroundJobWorker.run(AbstractBackgroundJobRunner.java:38)
	at org.jobrunr.server.runner.AbstractBackgroundJobRunner.run(AbstractBackgroundJobRunner.java:20)
	at org.jobrunr.server.BackgroundJobPerformer.runActualJob(BackgroundJobPerformer.java:95)
	at org.jobrunr.server.BackgroundJobPerformer.performJob(BackgroundJobPerformer.java:68)
	at org.jobrunr.server.BackgroundJobPerformer.run(BackgroundJobPerformer.java:46)
	at java.base/java.util.concurrent.ThreadPerTaskExecutor$TaskRunner.run(Unknown Source)
	at java.base/java.lang.VirtualThread.run(Unknown Source)
Caused by: org.axonframework.eventsourcing.eventstore.EventStoreException: An event for aggregate [775aba27-9cd9-4347-8ef5-353ad18c532d] at sequence [2] was already inserted
	at org.axonframework.eventsourcing.eventstore.AbstractEventStorageEngine.handlePersistenceException(AbstractEventStorageEngine.java:127)
	at org.axonframework.extensions.mongo.eventsourcing.eventstore.MongoEventStorageEngine.appendEvents(MongoEventStorageEngine.java:125)
	at org.axonframework.eventsourcing.eventstore.AbstractEventStorageEngine.appendEvents(AbstractEventStorageEngine.java:105)
	at org.axonframework.eventsourcing.eventstore.AbstractEventStore.prepareCommit(AbstractEventStore.java:66)
	at org.axonframework.eventhandling.AbstractEventBus.doWithEvents(AbstractEventBus.java:256)
	at org.axonframework.eventhandling.AbstractEventBus.lambda$null$11(AbstractEventBus.java:170)
	at org.axonframework.messaging.unitofwork.MessageProcessingContext.notifyHandlers(MessageProcessingContext.java:72)
	at org.axonframework.messaging.unitofwork.DefaultUnitOfWork.notifyHandlers(DefaultUnitOfWork.java:109)
	at org.axonframework.messaging.unitofwork.AbstractUnitOfWork.changePhase(AbstractUnitOfWork.java:236)
	at org.axonframework.messaging.unitofwork.AbstractUnitOfWork.commitAsRoot(AbstractUnitOfWork.java:87)
	at org.axonframework.messaging.unitofwork.AbstractUnitOfWork.commit(AbstractUnitOfWork.java:75)
	at org.axonframework.messaging.unitofwork.DefaultUnitOfWork.executeWithResult(DefaultUnitOfWork.java:95)
	at org.axonframework.messaging.unitofwork.UnitOfWork.executeWithResult(UnitOfWork.java:329)
	at org.axonframework.deadline.jobrunr.JobRunrDeadlineManager.execute(JobRunrDeadlineManager.java:246)
	... 10 common frames omitted
Caused by: com.mongodb.MongoCommandException: Command failed with error 251 (NoSuchTransaction): 'Transaction with { txnNumber: 244 } has been aborted.' on server production1-shard-00-04-pri.7udvj.mongodb.net:27017. The full response is {"errorLabels": ["TransientTransactionError"], "ok": 0.0, "errmsg": "Transaction with { txnNumber: 244 } has been aborted.", "code": 251, "codeName": "NoSuchTransaction", "$clusterTime": {"clusterTime": {"$timestamp": {"t": 1765803369, "i": 103}}, "signature": {"hash": {"$binary": {"base64": "StQor2wK4uWBjGP8G0rBFaJ5cds=", "subType": "00"}}, "keyId": 7534769034131145126}}, "operationTime": {"$timestamp": {"t": 1765803369, "i": 103}}}
	at com.mongodb.internal.connection.ProtocolHelper.getCommandFailureException(ProtocolHelper.java:205)
	at com.mongodb.internal.connection.InternalStreamConnection.receiveCommandMessageResponse(InternalStreamConnection.java:431)
	at com.mongodb.internal.connection.InternalStreamConnection.sendAndReceive(InternalStreamConnection.java:354)
	at com.mongodb.internal.connection.UsageTrackingInternalConnection.sendAndReceive(UsageTrackingInternalConnection.java:114)
	at com.mongodb.internal.connection.DefaultConnectionPool$PooledConnection.sendAndReceive(DefaultConnectionPool.java:743)
	at com.mongodb.internal.connection.CommandProtocolImpl.execute(CommandProtocolImpl.java:76)
	at com.mongodb.internal.connection.DefaultServer$DefaultServerProtocolExecutor.execute(DefaultServer.java:209)
	at com.mongodb.internal.connection.DefaultServerConnection.executeProtocol(DefaultServerConnection.java:115)
	at com.mongodb.internal.connection.DefaultServerConnection.command(DefaultServerConnection.java:83)
	at com.mongodb.internal.connection.DefaultServer$OperationCountTrackingConnection.command(DefaultServer.java:307)
	at com.mongodb.internal.operation.MixedBulkWriteOperation.executeCommand(MixedBulkWriteOperation.java:395)
	at com.mongodb.internal.operation.MixedBulkWriteOperation.executeBulkWriteBatch(MixedBulkWriteOperation.java:259)
	at com.mongodb.internal.operation.MixedBulkWriteOperation.lambda$execute$2(MixedBulkWriteOperation.java:203)
	at com.mongodb.internal.operation.SyncOperationHelper.lambda$withSourceAndConnection$0(SyncOperationHelper.java:124)
	at com.mongodb.internal.operation.SyncOperationHelper.withSuppliedResource(SyncOperationHelper.java:149)
	at com.mongodb.internal.operation.SyncOperationHelper.lambda$withSourceAndConnection$1(SyncOperationHelper.java:123)
	at com.mongodb.internal.operation.SyncOperationHelper.withSuppliedResource(SyncOperationHelper.java:149)
	at com.mongodb.internal.operation.SyncOperationHelper.withSourceAndConnection(SyncOperationHelper.java:122)
	at com.mongodb.internal.operation.MixedBulkWriteOperation.lambda$execute$3(MixedBulkWriteOperation.java:188)
	at com.mongodb.internal.operation.MixedBulkWriteOperation.lambda$decorateWriteWithRetries$0(MixedBulkWriteOperation.java:146)
	at com.mongodb.internal.async.function.RetryingSyncSupplier.get(RetryingSyncSupplier.java:67)
	at com.mongodb.internal.operation.MixedBulkWriteOperation.execute(MixedBulkWriteOperation.java:207)
	at com.mongodb.internal.operation.MixedBulkWriteOperation.execute(MixedBulkWriteOperation.java:77)
	at com.mongodb.client.internal.MongoClientDelegate$DelegateOperationExecutor.execute(MongoClientDelegate.java:173)
	at com.mongodb.client.internal.MongoCollectionImpl.executeInsertMany(MongoCollectionImpl.java:505)
	at com.mongodb.client.internal.MongoCollectionImpl.insertMany(MongoCollectionImpl.java:500)
	at com.mongodb.client.internal.MongoCollectionImpl.insertMany(MongoCollectionImpl.java:492)
	at java.base/jdk.internal.reflect.DirectMethodHandleAccessor.invoke(Unknown Source)
	at java.base/java.lang.reflect.Method.invoke(Unknown Source)
	at org.springframework.util.ReflectionUtils.invokeMethod(ReflectionUtils.java:281)
	at org.springframework.data.mongodb.SessionAwareMethodInterceptor.invoke(SessionAwareMethodInterceptor.java:119)
	at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:184)
	at org.springframework.aop.framework.JdkDynamicAopProxy.invoke(JdkDynamicAopProxy.java:223)
	at jdk.proxy3/jdk.proxy3.$Proxy201.insertMany(Unknown Source)
	at org.axonframework.extensions.mongo.eventsourcing.eventstore.AbstractMongoEventStorageStrategy.appendEvents(AbstractMongoEventStorageStrategy.java:95)
	at org.axonframework.extensions.mongo.eventsourcing.eventstore.MongoEventStorageEngine.lambda$appendEvents$2(MongoEventStorageEngine.java:122)
	at org.axonframework.common.transaction.TransactionManager.executeInTransaction(TransactionManager.java:47)
	at org.axonframework.extensions.mongo.eventsourcing.eventstore.MongoEventStorageEngine.appendEvents(MongoEventStorageEngine.java:121)
	... 22 common frames omitted

rsobies · December 21, 2025, 11:04am

I came to conclusion that mongo transactions are … problematic.
So I decided to get rid of it from AF, by providing empty implementation of TransactionManager

@Bean
fun createNonTransaction(): TransactionManager = TransactionManager {

        object : Transaction {
           
            override fun commit() {
            }

            override fun rollback() {
            }
        }
    }

Everything started to work as it should, no more writeConflict nor NoSuchTransaction