ReplayStatus always false and @DisallowReplay not working for PooledStreamingEventProcessor

[oysteing]

Basic information

• Axon Framework version: 4.5.7 (current giftcard-demo)
• JDK version:
• Complete executable reproducer if available (e.g. GitHub Repo): https://github.com/oysteing/giftcard-demo

Steps to reproduce

1. Start giftcard-demo modified in repo above
2. Issue single card and redeem card (2 events)
3. Initiate replay by pressing "Submit Bulk issue cards"

Expected behaviour

The expected behaviour is how the TrackingEventProcessor (axon.eventhandling.processors.card-summary.mode=tracking) works:

2022-03-25 11:37:17.268  INFO 24840 --- [nio-8080-exec-7] o.a.e.TrackingEventProcessor             : Shutdown state set for Processor 'card-summary'.
2022-03-25 11:37:17.269  INFO 24840 --- [nio-8080-exec-7] o.a.e.TrackingEventProcessor             : Processor 'card-summary' awaiting termination...
2022-03-25 11:37:17.622  INFO 24840 --- [card-summary]-0] o.a.e.TrackingEventProcessor             : Released claim
2022-03-25 11:37:17.623  INFO 24840 --- [card-summary]-0] o.a.e.TrackingEventProcessor             : Worker for segment Segment[1/1] stopped.
2022-03-25 11:37:17.651  INFO 24840 --- [card-summary]-1] o.a.e.TrackingEventProcessor             : Released claim
2022-03-25 11:37:17.651  INFO 24840 --- [card-summary]-1] o.a.e.TrackingEventProcessor             : Worker for segment Segment[0/1] stopped.
2022-03-25 11:37:17.665  INFO 24840 --- [card-summary]-2] o.a.e.TrackingEventProcessor             : Worker assigned to segment Segment[0/1] for processing
2022-03-25 11:37:17.666  INFO 24840 --- [card-summary]-2] o.a.e.TrackingEventProcessor             : Dispatching new tracking segment worker: TrackingSegmentWorker{processor=card-summary, segment=Segment[0/1]}
2022-03-25 11:37:17.666  INFO 24840 --- [card-summary]-2] o.a.e.TrackingEventProcessor             : Worker assigned to segment Segment[1/1] for processing
2022-03-25 11:37:17.666  INFO 24840 --- [card-summary]-3] o.a.e.TrackingEventProcessor             : Fetched token: ReplayToken{currentToken=null, tokenAtReset=IndexTrackingToken{globalIndex=1}} for segment: Segment[0/1]
2022-03-25 11:37:17.666  INFO 24840 --- [card-summary]-2] o.a.e.TrackingEventProcessor             : Using current Thread for last segment worker: TrackingSegmentWorker{processor=card-summary, segment=Segment[1/1]}
2022-03-25 11:37:17.667  INFO 24840 --- [card-summary]-2] o.a.e.TrackingEventProcessor             : Fetched token: ReplayToken{currentToken=null, tokenAtReset=IndexTrackingToken{globalIndex=1}} for segment: Segment[1/1]
2022-03-25 11:37:17.678  INFO 24840 --- [card-summary]-3] o.a.m.interceptors.LoggingInterceptor    : Incoming message: [CardIssuedEvent]
2022-03-25 11:37:17.679 DEBUG 24840 --- [card-summary]-3] i.a.d.g.query.CardSummaryProjection      : class io.axoniq.demo.giftcard.api.CardIssuedEvent handler called with replayStatus=REPLAY

Actual behaviour

Performing the same steps with PooledStreamingEventProcessor(axon.eventhandling.processors.card-summary.mode=pooled):

2022-03-25 11:38:51.969  INFO 24872 --- [nio-8080-exec-3] o.a.e.p.PooledStreamingEventProcessor    : Stopping processor [card-summary]
2022-03-25 11:38:51.985  INFO 24872 --- [nio-8080-exec-3] o.a.e.p.PooledStreamingEventProcessor    : Starting PooledStreamingEventProcessor [card-summary].
2022-03-25 11:38:51.986  INFO 24872 --- [card-summary]-0] o.a.eventhandling.pooled.Coordinator     : Processor [card-summary] claimed 2 new segments for processing
2022-03-25 11:38:51.994  INFO 24872 --- [card-summary]-1] o.a.m.interceptors.LoggingInterceptor    : Incoming message: [CardIssuedEvent]
2022-03-25 11:38:51.995 DEBUG 24872 --- [card-summary]-1] i.a.d.g.query.CardSummaryProjection      : class io.axoniq.demo.giftcard.api.CardIssuedEvent handler called with replayStatus=REGULAR
2022-03-25 11:38:51.995  INFO 24872 --- [card-summary]-1] o.a.m.interceptors.LoggingInterceptor    : [CardIssuedEvent] executed successfully with a [null] return value
2022-03-25 11:38:51.996  INFO 24872 --- [card-summary]-1] o.a.m.interceptors.LoggingInterceptor    : Dispatched messages: [CountChangedUpdate]
2022-03-25 11:38:52.001  INFO 24872 --- [card-summary]-1] o.a.m.interceptors.LoggingInterceptor    : Incoming message: [CardRedeemedEvent]
2022-03-25 11:38:52.002 DEBUG 24872 --- [card-summary]-1] i.a.d.g.query.CardSummaryProjection      : class io.axoniq.demo.giftcard.api.CardRedeemedEvent with @DisallowReplay handler called

The problematic part is the CardIssuedEvent handler called with replayStatus=REGULAR and the CardRedeemedEvent called although it has the @DisallowReplay annotation and a replay was performed.

[CodeDrivenMitch]

Hello @oysteing , thanks for reporting the issue with us! I have investigated and the ReplayToken is not properly propagated to the messages in the UnitOfWork when using a PooledStreamingEventProcessor. We are working on fix, which will be released in 4.5.9.

[smcvb]

Closing this issue, since it's resolved through pull request #2168.

[CodeDrivenMitch]

Hello @oysteing, release 4.5.9 has been released containing the fix. Good luck on your project!

[oysteing]

Thanks a lot for the quick fix and release. The fix seems to do the job. Will there also be a corresponding release of axon-bom?

[smcvb]

Sure thing @oysteing! I imagine a BOM release to come in today, by the way. So, stay tuned!

[smcvb]

The BOM's been released, @oysteing. It might take a little bit to get available, though.
I'm sure it'll be present tomorrow, though. :-)