Concurrent update in mongo

Axon Framework
, ,
1

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}}}

2

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?

3

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);
4

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.

5

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

6

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.