Events Propagation Issue

Axon Framework
1 
@EventHandler
    public void on(OrgProductPermissionAddedEvent event, MetaData metaData) {
        Long orgId = event.orgId();
        List<Long> productIds = event.productIds();
        List<OrgHumanBO> orgHumans = orgHumanGateway.getByOrgId(orgId);
        if (!CollectionUtils.isEmpty(orgHumans)) {
            orgHumans.forEach(e -> commandGateway.send(AddOrgHumanProductPermissionCommand.builder()
                    .orgHumanId(e.getId())
                    .productIds(productIds)
                    .permissionCustomize(PermissionCustomize.EXTENDS)
                    .build(), metaData));
        }
    }

Hello, I’d like to seek advice on a design issue: Here’s a snippet of my code where, while handling an event in the event handler, I send commands that affect other aggregate roots. This practice results in multiple events being recorded. During event replay, my OrgProductPermissionAddedEvent is executed once, and in the process, it sends the same commands again. However, since these events have been previously recorded, they also execute independently, leading to unintended exceptions. Is there a way to prevent such errors?

2

Hi Lius,

if you don’t want your handlers to be triggered during a replay, which is generally the case when you have side-effects in your handlers, annotate them with @DisallowReplay() or @AllowReplay(false).

3

Do note that commandGateway.send(...) returns a CompletableFuture. It would be wise to check it for the result of your command. Things might be failing without you noticing…

4

image
image2035×697 51.1 KB

very thanks,My approach was to reset the index to 0, allowing the event to re-execute,i’ll try to add the annotation for test

5

Setting the index to 0 won’t change anything, even with the annotation. Instead of manually manipulating the index, use the API on the Processor to reset it to the tail position.

6

:open_mouth:I wasn’t aware that this wasn’t a recommended approach; I merely proceeded based on my assumptions while testing the code… By the way, do u mind tell me how you usually record or handle the outcome of the commandGateway.send() method calls?

7

well, the send(...) method returns a CompletableFuture. This is essentially a way of saying “you’ll get the actual response later”. You can use the whenComplete(), thenApply(), etc methods to take action when the CompletableFuture completes.

In your case, since you’re in an event handler, you may want to send all commands, capture their CompletableFuture and combine them into a single one (e.g. using CompletableFuture.allOf(...)) to (block) wait for all of them to complete. You can also do .exceptionally() to decide what you want to do when an exception happens.
Unfortunately, Axon doesn’t yet support returning a CompletableFuture from an Event Handler method. Once the method returns, Axon moves on with the next handler/event.

Note that the use of CompletableFuture is not really Axon specific. I recommend checking out the Javadoc on it to find out how you use it to build efficient asynchronous code.

Alternatively, you can also use the commandGateway.sendAndWait(...) methods. This one blocks the sending thread, so the performance might be negatively impacted. It’s easier to read and write, though.

8

Hello, I have modified my code as follows:

CompletableFuture<ProductAggregate> exceptionally = queryGateway.query(new QueryProductCommand(productId),
                ResponseTypes.instanceOf(ProductAggregate.class)).exceptionally(throwable -> {
            throwable.printStackTrace();
            return null;
        });
        ProductAggregate productAggregate = null;
        try {
            productAggregate = exceptionally.get();
        } catch (Exception e) {
            //todo maybe send new Aggregate create command 
            throw new CommonException(ExceptionCode.COM_UNKNOWN);
        }

In my opinion, I want to resend the command for creating the aggregate and utilize the newly created aggregate root when the method fails to find the aggregate or an exception is thrown in the asynchronous thread. However, I sense this approach might entail numerous issues, such as encountering another exception during the recreation of the aggregate root. Are there any better strategies to address this scenario? Previously, I’ve also experimented with other APIs of queryGateway, including subscriptionQuery, but none have fulfilled my requirements. Does this mean my only option is to log the exceptions thrown by the command and resort to manual maintenance? Lastly, I genuinely appreciate your patience in answering so many of my queries.thank u so much!

9

It sounds like we’ve drifted away from the original question.

It might be best to start a new topic with a very specific question. Make sure to explain what you’re trying to achieve. That gives you the best chance of getting a useful answer.