Hi, we have an issue that we cannot seem to solve:
We want to upcast old events (e.g. created v0, updated v0) to a new version. The tricky part is, that in order to upcast e.g. the updated v0 event, we need information from previous events with the same aggregate id (e.g. the created v0 event.)
How would one do this in Axon?
We did try the ContextAwareSingleEventUpcaster, but the issue is that if we have something like the following to rebuild our read model, events from all aggregate ids are consumed, leading to an invalid read model.
@Service
class TransportOrderEventHandler(
private val readModelRepository: ReadModelRepository,
) {
@EventHandler
fun on(event: OurEvent) {
// Update the readModelRepository
// Apparantly the ContextAwareSingleEventUpcaster that upcasts "event" and passes it to this function, gets a stream of events not restricted to the event's aggregate id
}
}
abstract class JsonUpcaster(): ContextAwareSingleEventUpcaster<UpcasterContext>() {
public override fun canUpcast(
intermediateRepresentation: IntermediateEventRepresentation,
context: UpcasterContext
): Boolean {
// restrict to relevant event versions (e.g. "0" and type "created" or "updated")
}
public override fun doUpcast(
intermediateRepresentation: IntermediateEventRepresentation,
context: UpcasterContext
): IntermediateEventRepresentation {
// apply logic to upcast events. Fill the context.
// here we only want events for the same aggregate identifier
}
public override fun buildContext(): UpcasterContext {
return UpcasterContext()
}
}
First off, welcome to the forum, @anon26899999!
I hope your journey with Axon Framework has been pleasant so far.
Now, to the predicament at hand.
The ContextAwareSingleEventUpcaster will, as you’ve noticed, react on the specific Event Stream.
This means that for Event Sourcing, the stream is by nature limited to the aggregate identifier.
From the perspective of the Event Processors, the stream entails all events from a specified point in time (i.e. the token).
Making some assumptions around your sample, I have an idea we can discuss.
You can deal with this in the ContextAwareSingleEventUpcaster by having the buildContext object construct a collection from aggregateIdentifier to field-to-carry-over.
You would fill the context object when you’re hitting the event you need to carry data over from.
And you will thus read from the context to upcast the event that requires this data, by searching for the field based on the aggregate identifier.
Would this help you further, @anon26899999?
Thanks for your answer @Steven_van_Beelen .
That would be one option. However, can we be sure that the event stream contains the relevant events if there are a lot of others in between? (Assuming that our service does not restart in between, which we can prevent.)
In our case, imagine that the created event is from 2 month ago, and that before the next update happens on the same aggregate id, 1 million events were recieved in between.
However, can we be sure that the event stream contains the relevant events if there are a lot of others in between?
If the Event Processors starts early enough, you will be fine.
Thus, if you start from the beginning of time in your event stream, you’re ascertained you will pass the event.
So, taking your sample, if you would set up the upcaster and replay events and you know the gap between the carry-over event and the event-to-upcast is two months, then your replay should at least go back to two months ago.
Note that the upcasters are purely there for old versions of your events.
So, be certain that you actually adjust the event implementation so that new publications follow the desired format.
This makes it so that the upcasters are only required for old versions of your events.
Hence, the scope within which they’re useful will eventually die out, as most application work on recent data i.o. all data.
This is a rule of thumb, though. Perhaps your domain acts differently for how long your events are actively being used.