feat(pulsar): named-broker and broker-per-tenant support (GH-3308)

[jeremydmiller]

Closes #3308

Brings the Pulsar transport (Wolverine.Pulsar) up to parity with RabbitMQ / NATS / Kafka on the two multi-broker features it previously had neither of.

⚠️ Native-tenant vs. Wolverine tenant (the key design call)

Pulsar has its own native tenant segment in the topic hierarchy persistent://{tenant}/{namespace}/{topic} (the public in persistent://public/default/orders). Wolverine's broker-per-tenant "tenant" is an orthogonal Envelope.TenantId routing concept. This PR keeps the framework-consistent AddTenant(tenantId, …) name, but the per-tenant differentiator is a whole cluster connection (service URL + auth via IPulsarClientBuilder) — never the native tenant segment. Two Wolverine tenants normally use the same native tenant/namespace on different clusters. This distinction is called out prominently in the XML docs, the guide (a dedicated ::: warning), and PulsarTenant's summary.

Named broker

• PulsarTransport(string protocol) ctor (the parameterless ctor chains to it) so TransportCollection.GetOrCreate<T>(BrokerName) can build a named instance whose Protocol doubles as its endpoints' URI scheme.
• PulsarEndpointUri.Topic(topicPath, scheme) builds the endpoint URI from the transport Protocol instead of the hard-coded pulsar://; EndpointFor uses it. PulsarEndpoint.Parse already ignores the scheme.
• New public API: AddNamedPulsarBroker(BrokerName, Action<IPulsarClientBuilder>), ToPulsarTopicOnNamedBroker, ListenToPulsarTopicOnNamedBroker (applies the existing PulsarNativeResiliencyPolicy, mirroring UsePulsar).
• PulsarHealthCheck now reports the transport's own Protocol.

Broker-per-tenant (RabbitMQ-style: each tenant owns its own transport/client)

• New Internal/PulsarTenant — its own PulsarTransport/IPulsarClient, Compile(parent) copies DLQ/retry defaults. Because Pulsar connects in InitializeAsync (not ConnectAsync), each tenant client is built there and disposed in DisposeAsync.
• New public API on PulsarConfiguration: AddTenant(string, Action<IPulsarClientBuilder>), AddTenant(string, Uri serviceUrl) (convenience), TenantIdBehavior(TenantedIdBehavior).
• transport.BuildSender / BuildListenerAsync (modeled on RabbitMqTransport) fan out — when tenants are registered and the endpoint is TenantAware — to a TenantedSender and a CompoundListener (per-tenant PulsarListener wrapped in ReceiverWithRules + TenantIdRule). The hot-tail PulsarReaderListener branch is preserved per tenant. PulsarEndpoint.CreateSender/BuildListenerAsync delegate to these.

GH-2361

PulsarSender is fire-and-forget — it does not implement ISenderRequiresCallback — so it is safe under TenantedSender (no silent-drop regression). No Inline-sender switch was needed (unlike Kafka/SQS whose native senders use callbacks).

Tests

• CI-safe (run under CIPulsar): PulsarNamedBrokerTests (distinct transport per name, endpoint scheme == broker name, GetOrCreate<PulsarTransport>(name)), PulsarPerTenantConfigurationTests (AddTenant registers a PulsarTenant, Compile copies defaults, TenantAware endpoint resolves a TenantedSender, TenantIdBehavior setter), and PulsarNamedBrokerIntegrationTests — a real round-trip on the shared fixture broker proving the consumed envelope is stamped with the named broker's secondary:// scheme. 11/11 green.
• Local-only (excluded from CIPulsar): PulsarPerTenantConnectionTests ([Trait("Category","Integration")]) spins up a second Pulsar container as "cluster B" to prove true TenantId routing + inbound stamping. A deterministic single-broker tenant test is not feasible: Pulsar isolates tenants by cluster (unlike NATS's subject-prefix model), and two competing durable subscriptions on one broker collide — which is exactly why broker-per-tenant requires distinct clusters.

Build

dotnet build wolverine.slnx -c Release → 0 warnings, 0 errors.

Caveats

• The local-only two-cluster test was not fully validated end-to-end in this environment due to the known Testcontainers-Pulsar advertised-listener networking issue for a second container (the identical sender/listener code round-trips fine against the fixture broker in the CI-safe named-broker test); it will pass where a second Pulsar cluster is properly reachable.
• DotPulsar's client builder is not cloneable, so each AddTenant action runs against a fresh PulsarClient.Builder() and must fully specify ServiceUrl/auth (same "fresh factory" caveat as RabbitMQ).

🤖 Generated with Claude Code