ReplayStatus always false and @DisallowReplay not working for TrackingEventProcessor with parallelProcessing in multinode environment

[arth-of-war]

Basic Information

• Axon Framework 4.5.9
  • Replaying Tracking event processors, running parallel processing in multi node environment
• JDK version: 11

Spring Boot 2.5.6

We are running 2 instances of the application. Trigger manual replay via ReplayAPI,
EventHandler methods annotated with @DisallowReplay are still invoked and the ReplayStatus is REGULAR (should be REPLAY).

The same piece of code works fine in a single node environment.

Steps to Reproduce

Unfortunately I cannot replicate it locally and it is complicated for me to setup multiple nodes so I cannot specifically pinpoint the cause of the issue. But it may be something similar to this ticket but for TrackingEventProcessor and while running in a multi node environment?
#2154

1. Configuration:

eventProcessingConfigurer.registerTrackingEventProcessorConfiguration(config -> TrackingEventProcessorConfiguration
                .forParallelProcessing(2)
                .andInitialSegmentsCount(2)

4. Code snippet:

import static org.axonframework.modelling.command.AggregateLifecycle.apply;

import java.util.UUID;

import org.axonframework.commandhandling.CommandHandler;
import org.axonframework.commandhandling.gateway.CommandGateway;
import org.axonframework.eventhandling.AllowReplay;
import org.axonframework.eventhandling.DisallowReplay;
import org.axonframework.eventhandling.EventHandler;
import org.axonframework.eventhandling.ReplayStatus;
import org.axonframework.modelling.command.AggregateIdentifier;
import org.axonframework.spring.stereotype.Aggregate;
import org.springframework.stereotype.Component;

@Component
@AllowReplay
class EventHandlerWorking {
    private final CommandGateway commandGateway;

    public EventHandlerWorking(CommandGateway commandGateway) {
        this.commandGateway = commandGateway;
    }

    @EventHandler
    public void handle(ReplayingEvent event, ReplayStatus replayStatus) {
        System.out.println("ReplayStatus is correct. (REPLAY): " + replayStatus);
        commandGateway.send(new ReplayingCommand(UUID.randomUUID()));
    }
}

@Component
class EventHandlerNotWorking {
    /**
     * This should not be invoked
     */
    @EventHandler
    @DisallowReplay
    public void handle(DelegatedEvent event, ReplayStatus replayStatus) {
        // Status is expected as replay but it is REGULAR here
        System.out.println("ReplayStatus is wrong. Expected as REPLAY but REGULAR: " + replayStatus);
    }
}

class ReplayingEvent {
    private final UUID id;

    public ReplayingEvent(UUID id) {
        this.id = id;
    }
}

class DelegatedEvent {
    private final UUID id;

    public DelegatedEvent(UUID id) {
        this.id = id;
    }
}

class ReplayingCommand {
    private final UUID id;

    public ReplayingCommand(UUID id) {
        this.id = id;
    }

    public UUID getId() {
        return id;
    }
}

@Aggregate
class ReplayTest {
    @AggregateIdentifier
    private UUID id;

    @CommandHandler
    public void handle(ReplayingCommand command) {
        apply(new DelegatedEvent(command.getId()));
    }

    public ReplayTest() {
    }
}

Expected behaviour

EventHandlerNotWorking.handle() should not be invoked. ReplayStatus should be REPLAYING

Actual behaviour

EventHandlerNotWorking.handle() is still invoked and the ReplayStatus is REGULAR

[YvonneCeelie]

When running a replay did you stop one of the event processors?
Steps to do a replay:

1. Stop one of the 2 event processors
2. Start the replay on the other event processor
3. Start the event processor again

See https://docs.axoniq.io/reference-guide/axon-framework/events/event-processors/streaming#replaying-events

Resets in multi-node environments

[arth-of-war]

Yes @YvonneCeelie. I stopped all the processors in all instances before running the replay.

processor.shutDown();
processor.resetTokens();
processor.start();

I am now testing if using PooledStreamingProcessor is going to work, but I hope not to do this config change even if this works:

eventProcessingConfigurer
.usingPooledStreamingEventProcessors()
.registerTrackingEventProcessorConfiguration(config -> TrackingEventProcessorConfiguration
                .forParallelProcessing(2)
                .andInitialSegmentsCount(2)

Thanks!

When running a replay did you stop one of the event processors? Steps to do a replay:

1. Stop one of the 2 event processors
2. Start the replay on the other event processor
3. Start the event processor again

See https://docs.axoniq.io/reference-guide/axon-framework/events/event-processors/streaming#replaying-events

Resets in multi-node environments

[smcvb]

Note that the annotations only impact events that have been handled before.
So imposing a reset on an Event Processor that has not handled anything yet, would not mean no events are handled.

Having stated that, would you be able to show the token_entry table's state at the moment you trigger the reset?
If the tokens are, for example, at position 10 in the event stream, then the annotations only react to events 0 through 10.
All the events after that are regarded as "regular" event handling.

[arth-of-war]

You are correct @smcvb .
I was able to replicate it and the events regarded as REGULAR are new ones that were created as a side effect of the REPLAY.
Is there a way to propagate the ReplayStatus or ResetContext to the other command/event handlers in this case to know if the command was created from a REPLAY?

[smcvb]

Aaah gotcha, good to know you found this, @arth-of-war.

In most cases you wouldn't want tasks that introduce side-effects, like dispatching a command, to happen again during a replay.
I would wager the same applies to your setup.
Thus, instead of propagating the replay information to the command, to have the event handler not perform the command dispatching to begin with.

Would that solve the problem on your end, @arth-of-war?
If you feel that's not the case, would you mind sharing code snippets showing what you're doing exactly? That might give some guidance if the above doesn't help you.

[arth-of-war]

Hi @smcvb . Yeah I just added a condition to not dispatch the side effect commands during replay.
I was looking for a more generic approach to handle this scenario with a custom annotation and handler enhancers but it seems it is a bit complicated than I initially thought. Thanks!

[smcvb]

Thanks for your two cents, @arth-of-war.
Happy to hear the predicament is resolved.
As such, I'll close this ticket, removing the type "bug" in favor of "question."

Although I will close the issue, you should feel free to keep commenting here when applicable.