Axonframework: Upcaster leads to duplicated sequence-numbers

[pepperbob]

Today we detected an issue with the sequence number on event sourced Aggregates when using an upcaster:

removing events during the upcasting process, will lead to wrong computed sequence numbers when applying further events later on, e.g. for a specific aggregate my event store looks like this :

Global Index | Event Type | Sequence Number
-----| ---------- | ---------
1| EventA | 0
2| EventB | 1

Filtering out "EventB" like this:

@Override
protected Stream<IntermediateEventRepresentation> doUpcast(IntermediateEventRepresentation inter) {
        boolean droppingEvent = inter.getType().getName().contains("EventB")
        return droppingEvent ? Stream.empty() : Stream.of(inter);
    }

and sending a further command leading to "EventC" will leave me with an event store of:

Global Index | Event Type | Sequence Number
-----| ---------- | ---------
1| EventA | 0
2| EventB | 1
3| EventC | 1

It will consistently restart counting the sequence number from event that is filtered out.

If there is a unique index on the columns (aggregateidentifier, sequencenumber) all further commands will of course fail.

[smcvb]

Pretty sure this will only happen if the last event in the DomainEventStream when sourcing an Aggregate has been filtered out, as the "last known sequence number" of a given DomainEventStream is used to set the sequence number of an Aggregate Model.

I've done this type of event filtering before at users, but we never hit this potentially issue over there luckily.
We'll discuss if we can safe guard this in someway, but that might take some time.

Meanwhile, I'd suggest not to filter the event for now. Would you could do instead is keep the event but adjust the payload and payload type to something you're not (event) handling at all, like Void maybe. Then you'd still minimize the stream, but you'd keep the entry with it's sequence number, thus ensuring the EventSourcingRepository will still set the Aggregate's sequence number as it should.

[pepperbob]

I can confirm this happens with any event in the middle that's filtered out, too.

[smcvb]

Weird. That's definitely not my experience as I've already stated, and for now does not hold up with my observations.

The EventSourcedAggregate#initializeState(DomainEventStream) on line 280 calls the AnnotatedAggregate#initSequence(long) method which defines the sequence number used for subsequent published events after event sourcing said aggregate instance, using the DomainEventStream#getLastSequenceNumber method as input.
Looking further, it's the EventStreamUtils#upcastAndDeserializeDomainEvents method which is used to tie the upcaster you've written to the DomainEventStream. This utility even safes the sequence number of any entry, regardless of filtering.

Needless to say, I think I need a little more information about your project @pepperbob, as I fail to see where the problem lies at the moment.
Settings, storage solution, custom code you've written which ties in to this process, versions?

Even better, if you would be up for writing a test case which consistently shows this behavior, that would help tremendously.

[pepperbob]

Okay, this project shows the behaviour - it has a test that prints the sequences at the end.

https://github.com/pepperbob/super-duper-octo-succotash

[abuijze]

Hi @pepperbob , the tests in that project seem to confirm what Steven suggested. The problem occurs when sourcing an aggregate instance from its events. While reading, it keeps track of the largest sequence number it sees pass by. If the last event is removed from the stream, that event's sequence number will never be seen by the component at the end of it, and it will not know to update the sequence number.
We recommend to upcast to a "void" type, as Steven suggested, instead of removing an event entirely. I do realize that it would be nice if the framework could do that for you, but we just haven't gotten to that, yet.
You should be able to use the "void" payload type and an empty byte array of null value as the intermediate representation's value. That will generate enough information for Axon to work with, but it won't invoke any handlers (unless you have specified a catch-all handler accepting empty payloads).

--EDIT--
And we definitely recommend adding a unique constraint on the aggregateIdentifier, sequenceNumber combination! It helps ensure the consistency of all decisions recorded in the event store.

[abuijze]

If I change the implementation of your upcaster to this:

@Override
    public Stream<IntermediateEventRepresentation> upcast(Stream<IntermediateEventRepresentation> in) {
        return in.map(x -> x.getType().getName().equals(clazz.getName())
                          ? x.upcastPayload(SerializedType.emptyType(),
                                            byte[].class,
                                            b -> null)
                          : x);
    }

The sequence numbers are correctly increased.

[pepperbob]

@smcvb @abuijze Yes, just tinkered around with it a bit and it indeed only happens when the last event is filtered out - so hit a corner-case here.

Thanks for the hint with the SerializedType.emptyType - we already implemented a workaround to have at least a 1:1 ratio when upcasting events.

The Upcaster API is a bit misleading here - once we ran into that issue it got really hard to identify what's the actual problem and why there are unique constraint violations - effectively blocking the whole processor. I wonder if there's an easy way to calculate the sequence number before running the upcaster chain...

[smcvb]

I am uncertain how that exactly makes the Upcaster API misleading, @pepperbob.
Would you mind elaborating on that?

Or, even better, suggestions how to make it "not misleading" would be helpful.
Would a dedicated "filtering upcaster" resolved uncertainty in your scenario for example?

[pepperbob]

Obviously it seems it's not a major issue, however at the API allows you to manipulate the stream very effectively:

Stream<T> upcast(Stream<T> intermediateRepresentations);

Now, when filtering events you need to know that there might be a corner case of the "last event" - so filtering can basically not be recommended. The suggested workaround is fine, however it's not self-explanatory.

I even wonder where it misses the sequence number of the last event, as it looks as if EventStreamUtils already approaches to register the current sequence before applying the upcasters.

[smcvb]

Gotcha, the Upcaster API never refers to the suggested solution, ever. We'll tackle this one way or another eventually, hence why I feel this ticket should remain open.

Added, I get to the same conclusion as you hit @pepperbob when looking in to the EventStreamUtils class... I think the first step is to expand the EventStreamUtilsTest class with the issue you've hit to validate our assumption.

We're at the moment working on closing 4.3. After that release, I'll prioritize this issue to be picked up for either 4.4 or a bug release of 4.3.

[pepperbob]

I started looking at EventStreamUtils and saw that it does not have any issue. After poking around a while I came to the following conclusion:

The problem is, that every Implementation of DomainEventStream counts the sequence number while iterating over the Domain Events (i.e. after the upcaster ran) and therefore missing the potentially filtered "last event". The only exception is the anonymous implementation of DomainEventStream that's provided by its static method, see DomainEventStream:47

static DomainEventStream of(Stream<? extends DomainEventMessage<?>> stream, Supplier<Long> sequenceNumberSupplier)

This overwrites the getSequenceNumber() Method and links back to the Function provided by EventStreamUtils:93.

However, AbstractEventstore always returns a ConcatenatingDomainEventStream on readEvents(aggregateIdentifier) which does not delegate getSequenceNumber to the concatenated streams but calculate it on "its own" and therefore missing the filtered "last event", see ConcatenatingDomainEventStream:108.

To fix this ConcatenatingDomainEventStream should not rely on a locally stored sequence number but delegate it to the streams it concats, returning the highest sequence number found. Does that sound like a reasonable solution?

[abuijze]

Hi @pepperbob, that does indeed seem like a solid solution. Did you confirm that this change will resolve the filtering issue?

[pepperbob]

After those changes, my test case does not spit out any duplications anymore.

[smcvb]

Thanks as always for the effort @pepperbob, your fix will be part of the 4.3 release (which can be expected soon).