Fix global partitioning interceptor not intercepting Kafka messages

[jeremydmiller]

Summary

• Root cause: GlobalPartitionedInterceptor checked envelope.Message != null to decide whether to intercept, but Kafka envelopes arrive with only raw envelope.Data bytes — Message is always null at the receiver level because deserialization happens later in the handler pipeline. This meant the interceptor never fired for Kafka messages.
• Impact: In multi-node setups with global partitioning + Kafka, messages bypassed the companion local queues and were processed directly on whichever node received them. This caused concurrent access to the same Marten event stream across nodes, resulting in EventStreamUnexpectedMaxEventIdException.
• Fix: The interceptor now falls back to checking envelope.MessageType (the string type name from Kafka headers) when envelope.Message is null, deserializes via Pipeline.TryDeserializeEnvelope() before re-publishing, and GlobalPartitionedMessageTopology maintains a HashSet<string> of message type names for fast string-based matching.
• Test: Added a multi-node reproducing test (global_partitioned_aggregate_concurrency) that runs 2 Wolverine hosts with a separate publisher, confirming 0 concurrent access violations (was 17 before fix).

Test plan

• New multi-node concurrency test passes (0 concurrent access violations)
• All 6 existing global partitioning Kafka tests pass (no regressions)
• All 98 other Kafka tests pass (1 pre-existing flaky batch test excluded)
• All 1196 CoreTests pass

🤖 Generated with Claude Code