feat(kafka): broker-per-tenant multi-tenancy (dedicated cluster per tenant)

[jeremydmiller]

Closes #3303

Kafka already supported named brokers; this PR adds the missing broker-per-tenant feature. Each Wolverine tenant is served by its own dedicated Kafka cluster while sharing the topic topology, routed at runtime by Envelope.TenantId.

Approach

A Kafka "connection" is config-only (three Confluent config bags + builder callbacks), so — mirroring the RabbitMQ tenant model — each tenant owns a child KafkaTransport whose config is cloned from the parent and re-pointed at the tenant's own BootstrapServers (KafkaTenant.Compile). The child transport gets CreateProducer/Consumer/AdminClient for free and inherits EOS / idempotence / static-membership / SASL/SSL config.

Changes

• KafkaTenant (new) + KafkaTransport.Tenants cache. ConnectAsync compiles each tenant and, under AutoProvision, provisions the shared topics and the DLQ on every tenant cluster (each is an independent broker).
• KafkaTopic / KafkaTopicGroup fan out to a framework TenantedSender (outbound) and a CompoundListener with one KafkaListener per tenant cluster, stamping the tenant id inbound via ReceiverWithRules([new TenantIdRule(...)]). Dead-lettered messages are produced to the tenant's own cluster.
• New public API on KafkaTransportExpression: TenantIdBehavior(...), AddTenant(id, bootstrapServers), AddTenant(id, Action<KafkaTransportExpression>) (full auth/SASL/SSL, seeded from parent).
• Tenant/default senders under TenantedSender use fire-and-forget InlineKafkaSender. TenantedSender is intentionally not ISenderRequiresCallback (Message is stuck in outbox with multiple tenants and durable messaging #2361) and does not forward RegisterCallback, so a BatchedSender registered beneath it would never get its callback and would silently drop messages — inline production is the correct fire-and-forget model (same as the RabbitMQ / NATS tenant senders).

Consumer-group caveat

Because each tenant is a separate cluster with its own offsets, tenant listeners keep the same consumer GroupId — it is deliberately not suffixed per tenant (that would be wrong here, unlike NATS subject prefixing). Per-tenant retry topics / replay remain bound to the default cluster and are documented as out of scope.

Tests

• 6 CI-safe unit tests (KafkaPerTenantConfigurationTests, no broker): tenant compile clones config + re-points bootstrap servers, group id is not suffixed, parent config (SASL/idempotence/DLQ) inherited, Action-configured tenant seeded-from-parent-and-overridable, topics tenant-aware by default, a TenantAware endpoint resolves a TenantedSender.
• 3 integration tests (KafkaPerTenantConnectionTests, [Trait("Category","Integration")], second cluster via Testcontainers.Kafka, skips if Docker/broker unavailable): a DeliveryOptions{TenantId="tenantB"} message lands on the tenant cluster and not the default; a default message lands on the shared cluster; a tenant message round-trips back stamped with its TenantId.
• Full Kafka test project green (213 passed, 0 failed; pre-existing Flaky excluded). dotnet build wolverine.slnx -c Release clean (0 warnings, 0 errors).

Docs

New "Multi-Tenancy with a Broker per Tenant" section in kafka.md contrasting named-broker (static) vs tenant-routed (runtime DeliveryOptions.TenantId), with a compiling mdsnippets sample and the group-id / auto-provision caveats.

🤖 Generated with Claude Code