Background
During a recent load test, it was observed that, on rare occasions, certain events are not consumed by the saga as expected. This issue occurs during the execution of an operation that involves creating multiple aggregates and performing actions on them. Due to time constraints, the implementation was done in a slightly unconventional manner.
The operation begins with a command that then result in event that triggers the saga. Once started, all related commands are sent independently, and finally, a completion event is sent to end the saga. The entire process is handled within a single Unit of Work. If an error occurs during the process, a rollback mechanism triggers a compensation command. In cases where no error occurs but the operation fails to complete within the defined timeframe, the saga expires after reaching its deadline.
Observed Issue
Rarely, the SagaCompletedEvent sent during this operation is not consumed by the saga, resulting in the saga failing to complete as expected. Testing with brute-force changes seemed to mitigate the issue, which involved delaying the sending of the SagaCompletedEvent, but the exact root cause remains unclear. My hypothesis is that the SagaCompletedEvent is sent too quickly, before the saga is ready to handle this event, leading to situations where it expires after the deadline has passed.
For example, during a load test where 100 requests were sent to trigger this saga, in 2 cases the saga did not complete because the SagaCompletedEvent was not handled, even though it was sent according to the logs.
Below, I attach the pseudocode, logs, and other details that may help in diagnosing and resolving the issue.
@CommandHandler
public String handle(StartSagaCommand command) {
var id = command.id();
var compensationCommand = new CompensationCommand(id);
var unitOfWork = CurrentUnitOfWork.get();
commandGateway.sendAndWait(new SagaStartedCommand(id));
unitOfWork.onRollback(
message -> commandGateway.sendAndWait(compensationCommand));
unitOfWork.onPrepareCommit(message -> eventGateway.publish(new SagaCompletedEvent(id)));
executeCommands(command); //
return command.id();
}
@CommandHandler
protected void handle(SagaStartedCommand command) {
eventGateway.publish(new SagaStartedEvent(command.id()));
}
@Saga(sagaStore = "sagaStore")
@ProcessingGroup("sagaProcessor")
public class ExampleSaga {
public static final String DEADLINE_NAME = "DeadlineName";
private static final String ID_ASSOCIATION_KEY = "id";
@Autowired
private transient CommandGateway commandGateway;
@Autowired
private transient DeadlineManager deadlineManager;
private String id;
private String deadlineId;
@StartSaga
@SagaEventHandler(associationProperty = ID_ASSOCIATION_KEY)
public void on(SagaStartedEvent event) {
this.id = event.id();
deadlineId = deadlineManager.schedule(Duration.ofMinutes(10), DEADLINE_NAME, null);
}
@EndSaga
@SagaEventHandler(associationProperty = ID_ASSOCIATION_KEY)
public void on(SagaCompletedEvent event) {
deadlineManager.cancelSchedule(DEADLINE_NAME, deadlineId);
}
@EndSaga
@SagaEventHandler(associationProperty = ID_ASSOCIATION_KEY)
public void on(CompensationEvent event) {
executeCompensations();
deadlineManager.cancelSchedule(DEADLINE_NAME, deadlineId);
}
@DeadlineHandler(deadlineName = DEADLINE_NAME)
public void
onDeadline() {
executeCompensations();
SagaLifecycle.end();
}
private void executeCompensations() {
commandGateway.sendAndWait(new CompensationCommand(id));
}
}
2025-01-21T10:42:52.122Z DEBUG 1 --- [example-service] [nector(Axon)-97] c.e.c.EventDispatchInterceptor : Publishing event: [GenericEventMessage{payload={SagaStartedEvent[id=074a67e5-344a-441f-bc7f-8d58d110858e]}, ...}].
2025-01-21T10:42:52.195Z DEBUG 1 --- [example-service] [nector(Axon)-98] c.e.c.EventDispatchInterceptor :
Publishing event: GenericEventMessage{payload={SagaCompletedEvent[id=074a67e5-344a-441f-bc7f-8d58d110858e]}, ...}].
2025-01-21T10:42:53.091Z INFO [example-service] [Processor]-0] c.e.saga.ExampleSaga:
ExampleSaga(Id=074a67e5-344a-441f-bc7f-8d58d110858e): Scheduled discontinue deadline=deadline-8d155966-02de-4ddf-a86f-a516a5e827b0.
2025-01-21T11:42:53.094Z WARN [example-service] [Scheduler_Worker-5] c.e.saga.ExampleSaga:
ExampleSaga(Id=074a67e5-344a-441f-bc7f-8d58d110858e): Deadline 'DeadlineName' reached. Executing compensating actions.
2025-01-21T11:42:53.094Z INFO [example-service] [Scheduler_Worker-5] c.e.saga.ExampleSaga:
ExampleSaga(Id=074a67e5-344a-441f-bc7f-8d58d110858e): Executing compensating actions.
eventhandling:
processors:
sagaProcessor:
mode: tracking
source: streamableKafkaMessageSource