Resolve incorrect event routing in heterogeneous Sequencing Policy setup among Event Handling Components

[MateuszNaKodach]

Route events per component when handling across segments

Summary

A PooledStreamingEventProcessor could handle the same event more than once and in the
wrong segment whenever a single processor hosted multiple EventHandlingComponents that use
different SequencingPolicy instances. This PR makes event handling apply the same
per-component segment routing that the admission filter already relies on, so each component
handles an event exactly once, in the single segment it belongs to.

Background: how a pooled processor routes an event

A PooledStreamingEventProcessor runs one WorkPackage per Segment, and every work package
shares the same set of EventHandlingComponents (wrapped in ProcessorEventHandlingComponents).
Which segment an event belongs to is derived from a component's sequence identifier
(SequencingPolicy.sequenceIdentifierFor(...)), hashed and masked against the segment.

Crucially, routing is decided in two separate phases:

sequenceDiagram
    participant Coord as Coordinator
    participant WP as WorkPackage for Segment S
    participant Filter as DefaultWorkPackageEventFilter
    participant PEHC as ProcessorEventHandlingComponents

    Note over Coord,Filter: Phase 1 - Scheduling / admission
    Coord->>WP: scheduleEvent(E)
    WP->>Filter: canHandle(E, ctx, Segment S)
    Filter-->>WP: boolean - does ANY component identifier hash to S
    Note over WP: result stored per entry, E enqueued only if true

    Note over WP,PEHC: Phase 2 - Handling, later, per batch
    WP->>PEHC: handle(batch, ctx) with Segment S in context
    Note over PEHC: BEFORE - runs EVERY supporting component, segment ignored
    Note over PEHC: AFTER - runs only components whose identifier hashes to S

Loading

The important detail: admission produces a single boolean per (segment, event). It answers
"does this event belong in this segment for anybody?" using anyMatch over all component
identifiers. It does not record which component caused the match.

The problem

Each EventHandlingComponent may declare its own SequencingPolicy. So for one and the same
event, component A and component B can resolve different sequence identifiers and therefore
legitimately belong to different segments. This is a supported topology, not a
misconfiguration.

Before this change:

• Admission (DefaultWorkPackageEventFilter.canHandle) admitted the event into a segment's
  queue when any component's identifier hashed into that segment.
• Handling (ProcessorEventHandlingComponents.handle) then invoked every component that
  supports(...) the event and never consulted the segment - even though the Segment is sitting
  in the ProcessingContext under Segment.RESOURCE_KEY.

Because the per-component reason for admission was discarded, every admitting segment re-ran all
components:

flowchart TD
    E["Event E<br/>A -> hashes to segment 0<br/>B -> hashes to segment 1"]

    E --> F0
    E --> F1

    subgraph S0["WorkPackage - Segment 0"]
        F0{"Filter: any identifier hashes to seg 0?<br/>YES (via A)"}
        H0["handle(): run ALL supporting components"]
        A0["A handles E   (correct)"]
        B0["B handles E   (WRONG - B belongs to seg 1)"]
        F0 -->|admitted| H0
        H0 --> A0
        H0 --> B0
    end

    subgraph S1["WorkPackage - Segment 1"]
        F1{"Filter: any identifier hashes to seg 1?<br/>YES (via B)"}
        H1["handle(): run ALL supporting components"]
        A1["A handles E   (WRONG - A belongs to seg 0)"]
        B1["B handles E   (correct)"]
        F1 -->|admitted| H1
        H1 --> A1
        H1 --> B1
    end

Loading

Net result: A is handled twice and B is handled twice (once per segment) instead of exactly
once each. This duplicates side effects (projections, notifications, mailings, ...) and breaks the
single-writer-per-sequence guarantee the sequencing policy is meant to provide.

Why it stayed hidden

In the common configuration every component yields the same identifier for a given event
(e.g. all sequence by aggregate id, or all use the default policy). Then sequenceIdentifiersFor
collapses to a single-element set that hashes to exactly one segment: only that segment admits the
event and all components run once. No duplication. The defect only appears once components use
heterogeneous sequencing policies for the same event - which is exactly the case that was
untested.

The fix

Handling now applies the same per-component routing decision the filter already relies on: a
component handles an event only when its own sequence identifier hashes into the segment currently
attached to the context.

flowchart TD
    E["Event E<br/>A -> hashes to segment 0<br/>B -> hashes to segment 1"]

    E --> F0
    E --> F1

    subgraph S0["WorkPackage - Segment 0"]
        F0{"Filter: any identifier hashes to seg 0?<br/>YES (via A)"}
        H0["handle(): canSegmentHandle(seg0, component) per component"]
        A0["A: id hashes to 0 -> HANDLES E"]
        B0["B: id hashes to 1 -> skipped"]
        F0 -->|admitted| H0
        H0 --> A0
        H0 --> B0
    end

    subgraph S1["WorkPackage - Segment 1"]
        F1{"Filter: any identifier hashes to seg 1?<br/>YES (via B)"}
        H1["handle(): canSegmentHandle(seg1, component) per component"]
        A1["A: id hashes to 0 -> skipped"]
        B1["B: id hashes to 1 -> HANDLES E"]
        F1 -->|admitted| H1
        H1 --> A1
        H1 --> B1
    end

Loading

Net result: A once, B once - each in the single segment it belongs to. Admission and handling
now agree.

Concretely, in ProcessorEventHandlingComponents:

1. Per-component segment gate while handling. handle(...) reads the Segment from the
   context (Segment.fromContext) and invokes a component only when it supports the event and
   canSegmentHandle(segment, component, event, context) is true. The check is
   segment.matches(component.sequenceIdentifierFor(event, context)) - the identical computation
   the filter performs, so the two phases can no longer diverge.

2. Non-segmented fallback preserved. When no Segment is attached to the context (for example
   a SubscribingEventProcessor), canSegmentHandle returns true and every supporting component
   handles the event, exactly as before.

3. sequenceIdentifiersFor(...) now ignores non-supporting components. A component that does
   not support the event no longer contributes an identifier, so it can no longer cause a segment
   to claim an event it will never handle. This keeps the admission and handling views of "which
   components route here" symmetric.

DefaultWorkPackageEventFilter, WorkPackage, Segment and SegmentMatcher are unchanged.

Testing

Added, following TDD (written to fail against the previous behaviour first):

Unit tests (ProcessorEventHandlingComponentsTest -> @Nested SegmentAwareRouting)

• an event handled within a segment only invokes the component whose identifier routes to that
  segment (one test per segment);
• with no segment in the context, every supporting component handles the event;
• sequenceIdentifiersFor(...) excludes components that do not support the event.

End-to-end tests (PooledStreamingEventProcessorTest -> @Nested SegmentRouting, initialSegmentCount(2))

• two components with different identifiers each handle the event exactly once, in their own
  segment;
• components supporting different event types each handle only their own type, exactly once;
• two components sharing a sequence identifier both handle the event once in the same segment
  (guards against over-restriction).

Verification:

• Temporarily reverting the production fix turns the relevant unit tests and 2 of the 3 end-to-end
  tests red (duplicate deliveries), confirming they are genuine regression guards.
• Full messaging module: 4139 tests, 0 failures, 0 errors (505 skipped).

Compatibility

• Single-segment processors (default, mask == 0) are unaffected: Segment.matches always
  returns true, so every supporting component still handles every supported event.
• SubscribingEventProcessor and any non-segmented handling path are unaffected (no Segment in
  context -> all supporting components run).
• Behaviour changes only for multi-segment pooled processors whose components use different
  sequencing policies - the case that was previously duplicating deliveries.

[hatzlj]

nice catch 👍, no blocking issues for me

[hatzlj]

Does it probably make sense to reuse the SegmentMatcher logic here or create a shared helper, to make sure the same logic as for scheduling (as you state in the Javadoc) is invoked in both places.
I mean, the logic is trivial, but as you point out in the javadoc for this method, it should "mirror the routing decision made while scheduling the event".

[smcvb]

Liking this suggestion too.

[hatzlj]

Suggested change

	private static boolean canSegmentHandle(@Nullable Segment segment,
	private static boolean canHandleInSegment(@Nullable Segment segment,

[smcvb]

I prefer this rename suggestion

[smcvb]

Once the two nits from @hatzlj have been tackled, you're free to merge this in, @MateuszNaKodach!

Ow, and I think we should port this fix to axon-5.2.x as well, so that it'll be a part of 5.2.1. Thus, would you mind following this up with a port PR once done?

[smcvb]

I prefer this rename suggestion

[smcvb]

Liking this suggestion too.