GH-3533: Pin Azure Service Bus session listeners to specific session identifiers

[jeremydmiller]

Closes #3533

Thanks @tudor-tesss for the well-scoped request — the mechanism you described (session id as a broker-enforced routing key) is exactly what this implements.

Problem

There was no way to have several competing consumers share a single Azure Service Bus queue/subscription while each only ever processes messages for a fixed set of session ids. The hand-rolled session listener always calls AcceptNextSessionAsync(), so a consumer locks whatever session is next available — the cross-delivery problem from the issue.

Solution

Add an opt-in ServiceBusSessionProcessor-based session listener and expose ServiceBusSessionProcessorOptions to users. Populating SessionIds pins the processor so it only ever locks those sessions; on a shared entity that makes the session id a broker-enforced routing key.

API

// Consumer node "A" — only ever locks the "A" session on a shared queue
opts.ListenToAzureServiceBusQueue("shared-orders")
    .RequireSessionsWithOnlyTheseIdentifiers("A");

// Consumer node "B"
opts.ListenToAzureServiceBusQueue("shared-orders")
    .RequireSessionsWithOnlyTheseIdentifiers("B");

// Producer routes with the session id (GroupId)
await bus.PublishAsync(new OrderPlaced(...), new DeliveryOptions { GroupId = "A" }); // only A receives

The general hook for the other knobs:

opts.ListenToAzureServiceBusQueue("orders")
    .ConfigureSessionProcessor(o =>
    {
        o.MaxConcurrentSessions = 8;
        o.MaxAutoLockRenewalDuration = TimeSpan.FromMinutes(10);
        o.SessionIdleTimeout = TimeSpan.FromSeconds(30);
    });

Both methods are available on queue and subscription listener configs.

What's in the box

• ConfigureSessionProcessor on AzureServiceBusEndpoint — a multicast delegate (unlike inline ConfigureProcessor) so the SessionIds sugar and explicit customizations compose.
• BuildSessionProcessorOptions mirrors BuildProcessorOptions: maps ListenerCount → MaxConcurrentSessions, keeps in-session FIFO (MaxConcurrentCallsPerSession = 1), and re-asserts the acknowledgement-critical ReceiveMode = PeekLock / AutoCompleteMessages = false.
• New InlineAzureServiceBusSessionListener (built on ServiceBusSessionProcessor) with full ISupportDeadLetterQueue / ISupportNativeScheduling / IReportConnectionState parity.
• AzureServiceBusEnvelope gains a ProcessSessionMessageEventArgs complete/defer/dead-letter path (it does not share a base type with ProcessMessageEventArgs).

Safety / gating

The existing hand-rolled AcceptNextSession loop remains the default. The processor engine is only used when ConfigureSessionProcessor is set (directly or via RequireSessionsWithOnlyTheseIdentifiers). Existing session listeners are completely unaffected — this is purely additive. Wolverine reserves ReceiveMode and AutoCompleteMessages.

Tests

• Unit coverage for BuildSessionProcessorOptions and both fluent methods (SessionIds population, ListenerCount→MaxConcurrentSessions, PeekLock/AutoComplete re-assertion, multicast composition, and that a plain RequireSessions() still uses the legacy loop).
• An emulator end-to-end test that reproduces the issue: a listener pinned to "A" receives exactly its messages, while a "B" message seeded onto the same shared queue is never delivered to it.

Docs added to the Azure Service Bus session-identifiers page + DocumentationSamples.

🤖 Generated with Claude Code

https://claude.ai/code/session_01FKAxzuZ36VP6UPcTQf3MUs