As the event could result from a replay of events that were created in the past, I need the ‘sagaStarted’ field to reflect that original creation time of the event. However, I’m getting the time that the event was applied to an aggregate.
I suspect this is a result of how we handle a replay of events. What we’re doing is deserialising events received, batching them up in to a command message, and then issuing that command containing - say - 500 events. The command handler then iterates over that collection of events applying each one to the aggregate.
Having a look though the code, it seems events delivered to the application arrive with the expected date and time, but then when those events are applied to the aggregate in a @CommandHandler method, the apply() method goes through a number of calls which ultimately result in the creation of new GenericDomainEventMessage<> which does not pick the timestamp out of the original event message, but just initialises it with a new DateTime instance, therefore losing the original events creation time. When this event is delivered to the Saga, I get the time the event was applied to the aggregate, not the time the event was created.
I suspect if this event was delivered directly from the event bus at the point it was received, this wouldn’t happen. But applying it to the aggregate first and having that publish the event it causing this to happen.
How can I avoid this from happening? Is apply() the correct call to be making, or should I call registerEventMessage() directly??
It’s very irregular to apply old events to an aggregate via a command as a way of pumping existing events into your application. From your question I’m guessing that you need the events (from another Axon node?) to be added to an event store before replaying and that you’re (ab)using an aggregate to do this. If these events are not in an event store yet you can add them to an event store yourself without the use of an aggregate before you initiate the replay.
Looking at your code it appears you’re still on Axon 2. What you could do in Axon 2 is use a ReplayingCluster containing the right event handlers (in this case your saga) and initiate the replay. If you only want to replay events from a single aggregate you can select events by aggregate id.
It’s unclear to me why you are handling events in batches of 500 like this. The whole operation of batching events seems iffy to me and my email is probably at best solving an issue that is not at the heart of the problem :).
Thanks for that - it’s a fair question and I’ll ask the team why we do this. I have a vague feeling it was to work around some other issues we had as I don’t think we always handled events in batches like this…
But yes, we are on Axon 2 and the scenario in which this happens is when a node in a cluster starts up completely clean (no event store). It connects to a number of other applications and creates an aggregate for each of them, applying the events they generated in their lifetime to rebuild the current view of them.
Although this is probably just a continuation of the abuse - I just tried regsiterEventMessage() instead of apply() and I get the correct timestamp delivered to the saga, however I’m not clear if these events are applied to the aggregate. The JavaDoc states apply() adds the event to the uncommitted event queue, but registerEventMessage() doesn’t, it just says it’s published after the aggregate is saved. Does this mean that registerEventMessage() bypasses the mechanism that applies events to the aggregate and just stores them up and sends them when the command handler completes?
Hmm I believe those events ARE added to the store but are not applied to the aggregate the first time :). I would say, just check if they’re in the store…
the registerEvent doesn’t apply the message to the aggregate. It’s just the part of the process that stages it for publication when the Unit of Work is committed.