Axonframework: isCaughtUp() of TrackingEventProcessor always returns true

[bastoker]

In our production situation the isCaughtUp property of the EventTrackerStatus for all segments of an TracingEventProcessor always render to true. Even at startup of application when (in our situation) we start with empty TrackingToken and hence start at tail of event stream.

Context

We use Axon Server version 4.2 and have a separate SpringBoot microservice that has an event listener (using axon-spring-boot-starter).
In this microservice we use an injected EventProcessingConfiguration we execute the following code once in a while, to see if our microservice is ready to serve our clients:

val eventProcessorOption = procConfig.eventProcessor(eventProcessorName, TrackingEventProcessor::class.java)
if (eventProcessorOption.isEmpty) {
    throw IllegalArgumentException("Event Processor with name $eventProcessorName and type 'TrackingEventProcessor' not found")
}

val trackerStatusPerSegment = eventProcessorOption.get().processingStatus()
 if (trackerStatusPerSegment.isEmpty()) {
    AvailabilityChangeEvent.publish(eventPublisher, this, ReadinessState.REFUSING_TRAFFIC)
    return;
 }

val isCaughtUp: Boolean = trackerStatusPerSegment.entries.stream().allMatch { it.value.isCaughtUp }
...
// isCaughtUp is always true when microservices is started (with clean token store so it will always read all events starting at tail of stream

Research

I did some debugging and my initial feeling is that the following call in TrackingEventProcessor#processBatch()
to checkSegmentCaughtUp(segment, eventStream) should include a timeout as well, just like the call to eventStream.hasNextAvailable(...)?

private void processBatch(Segment segment, BlockingStream<TrackedEventMessage<?>> eventStream) throws Exception {
        List<TrackedEventMessage<?>> batch = new ArrayList<>();
        try {
            checkSegmentCaughtUp(segment, eventStream);
            TrackingToken lastToken;
            Collection<Segment> processingSegments;
            ...

The call to checkSegmentCaughtUp() is responsible for setting the isCaughtUp flag to true.
It calls this method to check, but this has a hard coded waiting interval of 0 seconds. Maybe this should be configurable just like e.g. EventAvailabilityTimeout

public interface BlockingStream<M> extends AutoCloseable {
    /**
     * Checks whether or not the next message in the stream is available. If so this method returns
     * {@code true} immediately. If not it returns {@code false} immediately.
     *
     * @return true if a message is available or becomes available before the given timeout, false otherwise
     */
    default boolean hasNextAvailable() {
        try {
            return hasNextAvailable(0, TimeUnit.NANOSECONDS);
        } catch (InterruptedException e) {
            return false;
        }
    }
    ...

I am very curious if this is due to a config error or something else, but we tried different settings like batch-sizes and timeouts and this issue is reproducible (connecting to our server) with axon framework versions 4.2, 4.2.2 and 4.3.5.

References

This issue is already mentioned, a little while ago, at the mailing list. See https://groups.google.com/g/axonframework/c/kfc3donlVAQ/m/0wwQ7Nr8BwAJ

[smcvb]

Thanks for the detailed report here @bastoker.
Always happy to see the viciousness people have to figuring out an issue.

However, let me share the JavaDoc of the EventTrackerStatus#isCaughtUp method with you:

/**
 * Whether the Segment of this status has caught up with the head of the event stream. Note that this is no
 * guarantee that this segment is still processing at (or near) real-time events. It merely indicates that this
 * segment has been at the head of the stream since it started processing. It may have fallen back since then.
 *
 * @return whether the Segment of this status has caught up with the head of the event stream
 */

What this piece tells us is that the isCaughtUp flag is intended to only be set once. The only time it's changed is when the head of the stream is reached for the first time. With an empty event stream, this would mean it's true from the get go, as it would've reached the head. For an event stream of n, this would only become true once all events at that point in time have been handled.
The head is a moving target though, so at any moment in time the event stream could receive _N_ additional events. This thus _does not_ influence the caughtUp flag once it has been set to true.

Simply put, a Segment (read: part of a TrackingToken) has reached the end of the stream once, it will always show as caughtUp.

There's one issue with this explanation of mine which doesn't make it a full reasoning for what you're seeing though:

Even at startup of application when (in our situation) we start with empty TrackingToken and hence start at tail of event stream.

What's currently unclear to me is that this behaviour is something you've tested with an entirely clean set up yes or no.
If yes, I am assuming there are no events. Thus, the event stream would've reached the end immediately, resulting in caughtUp == true from the start. If this is not the case though...well I'd need to have a think in that case. :-)

So, if you could tell me whether you spot this behaviour with a non-empty event store.

[bastoker]

Hi Steven,
We indeed spot this behaviour with a non-empty event store, as of the get-go.
There is a lot of Spring config magic going on at our setup at the moment, so I first try a more vanilla approach as of the config goes, and if this behaviour still occurs I'll try to create a fork to reproduce this in e.g. a unit test.

[smcvb]

Good to know Bas.
Then I guess we are spotting default behavior here.

I know this will require this microservice to have a different hook to tie into when it can actually start it's work of course.
What you could do in that case is validate the size of the event store against the position of your tracking tokens. If that is the same value, then you can be certain your event processing threads are at the head of the stream at that point in time. Chances are however high this is a moving target, so utilizing a window of opportunity here might be better suited.

Regardless, like to hear from you what your following investigation will bring!

[bastoker]

What do you mean with default behaviour? I said non-empty as opposed to an empty event store, so I don't get what you mean by that ;-)

[smcvb]

Sorry for the radio silence here @bastoker. We've had a short internal discussion and noticed you did find something which shouldn't occur. My thought of "default behaviour" was in the case a non-AxonServer based solution was being used, which isn't the case in your sample. I've provided a fix for the issue, which doesn't introduce a new setting but in essence uses the eventAvailabilityTimeout. The issue number for the PR is #1512.

So...thanks for filing this with us @bastoker! Not sure when we would've noticed this otherwise.

[bastoker]

This is awesome news, thanks for solving this!

[smcvb]

Closing this issue as it has been resolved in pull request #1512. It'll be included in 4.4.3 @bastoker, which I aim to release tomorrow. Stay tuned :-)

[bastoker]

@smcvb If I want to test this, can I just upgrade the axon-framework in my event sourcing-client and keep my axon-server as is (currently running version 4.2.2)?

[smcvb]

It should be perfectly fine to keep Axon Server on 4.2.2 and move the framework over to 4.4.3 :-)

[maartenjanvangool]

So in the PR you added

checkSegmentCaughtUp(segment, eventStream);

in a few places.

As far as I can tell, this does not do this:

I've provided a fix for the issue, which doesn't introduce a new setting but in essence uses the eventAvailabilityTimeout

private void checkSegmentCaughtUp(Segment segment, BlockingStream<TrackedEventMessage<?>> eventStream) {
  if (!eventStream.hasNextAvailable()) {
    updateActiveSegments(() -> activeSegments.computeIfPresent(segment.getSegmentId(), (k, v) -> v.caughtUp()));
  }
}

with the hasNextAvailable using the default parameters:

hasNextAvailable(0, TimeUnit.NANOSECONDS);

and therefore does not use the eventAvailabilityTimeout.

Shouldn't the checkSegmentCaughtUp take the event availability timeout into account as well? As far as I can tell, pr #1512 does not resolve the problem @bastoker encountered.

[abuijze]

The "trick" is that the checkSegmentCaughtUp call is now done only after waiting for events to process. Only after a batch has been accumulated, the call is done to validate if we are at the head of the stream.