[christopherread]

Fixes the two open Obvs.Kafka items under "Kafka on managed platforms" in #130, both found while building #135.

1. Transient errors no longer end subscriptions

Before this change, every error the Kafka client reported went to observer.OnError, transient ones too, such as "1/1 brokers are down" or a connection the platform closed. An unreadable message also ended the subscription.

How the client reports errors, from its own documentation:

• The error handler (SetErrorHandler): Confluent.Kafka's docs say "the client will try to automatically recover from errors that are not marked as fatal", and that non-fatal errors are informational. librdkafka's INTRODUCTION.md says the same about error_cb. For a consumer, fatal errors come from static group membership being fenced, and the consumer stops itself afterwards.
• ConsumeException from Consume: a consumer error. Most are reported once and the client carries on, for example an unknown topic, or a message that fails deserialization (which has already been consumed). librdkafka says topic authorization failures are reported "once per partition" and the fetcher backs off. The application should act on them, because they don't fix themselves.
• A failed OAUTHBEARER token refresh arrives as Local_Authentication. librdkafka keeps the token it has and retries in 10 seconds (rd_kafka_oauthbearer_set_token_failure0 in rdkafka_sasl_oauthbearer.c). So authentication errors are treated as transient: ending the subscription on one hiccup at the token endpoint would be wrong. Bad credentials are already caught at startup, where KafkaTopics.EnsureCreated fails straight away.

So MessageSource now ends the subscription, with a KafkaException carrying the client's Error, only for:

• Error.IsFatal;
• TopicAuthorizationFailed, GroupAuthorizationFailed and ClusterAuthorizationFailed.

When one of these ends the subscription, the consumer also stops waiting in Consume. Everything else is logged as a warning in the Obvs.Kafka category, from the host's ILoggerFactory when using UseKafka, and consumption carries on.

2. Consumer groups are named after the service, not the process

How it worked before: v6's fluent configuration put every source in one group, <machine>-<user>-<pid>. UseKafka used <machine>-<user>-<pid>-<topic>-<instance ID> for every topic. So with UseKafka, commands and requests were not shared between a service's buses: every bus handled every command. The existing test ShouldDeliverToEveryBusInTheProcess asserted that. And every restart created new groups.

Now, with UseKafka:

Topic	Group
<service>.Commands	<service>.Commands, shared by the service's buses, so each command is handled once
<service>.Requests	<service>.Requests, shared
<service>.Events	<service>.Events.<instance ID>, one per bus, so every bus gets every event
<service>.Responses	<service>.Responses.<instance ID>, one per bus

Each group name starts with <service>., so access control can grant a prefix instead of *. The Confluent Cloud and MSK pages now say so.

Sharing a group needs one more change. A new group's start position (the end of the log) is now committed when partitions are assigned. Without that, a bus that takes over a partition from one that consumed nothing would skip to the end again, missing what was sent in between. A restarted bus would do the same. If that commit fails, the error is logged and the consumer still starts from that position.

v6's WithKafkaEndpoints<...>() keeps its process-named groups. KafkaServiceEndpointProvider's uniqueGroupSuffix parameter becomes instanceId, and MessageSource's becomes groupId. Both were added in v7 (#119) and aren't in a release.

Behaviour change: ShouldDeliverToEveryBusInTheProcess asserted the old behaviour, so I replaced it with ShouldHandleEachCommandOnceAcrossTheServicesBuses and ShouldDeliverEveryEventToEveryBus. This is the change the task asked for, so the old test wasn't wrong when it was written.

Docs

• docs/transports/kafka.md has new sections: "Consumer groups" (the naming, partitions limiting how far work is shared, UseInstanceId(...) for stable groups, and groups left behind by buses with random instance IDs, which Kafka removes offsets.retention.minutes after they empty, 7 days by default) and "Errors while running". The Permissions section now gives the group prefix.
• kafka-azure-event-hubs.md: the limit of 1,000 Kafka consumer groups per namespace, how many groups each service and bus uses, and the advice to use a fixed instance ID per bus. Microsoft doesn't document when Event Hubs removes unused Kafka groups, and the page says so.
• kafka-confluent-cloud.md, kafka-amazon-msk.md: prefixed group permissions.

Testing

Test-driven: each test below was written first and seen failing for the expected reason before the implementation.

• TestMessageSourceErrors, unit tests:
  • 7 errors the client recovers from are logged and consumption carries on. They failed with Assert.False because every error ended the subscription.
  • 5 fatal or authorization errors end the subscription with a KafkaException. They failed because a plain Exception was passed.
• TestMessageSourceErrors.ShouldSkipAMessageItCantReadAndCarryOn (Explicit) failed because a ConsumeException: Local: Value deserialization error ended the subscription.
• TestKafkaConsumerGroups, unit tests: the share and per-bus tests failed because groups were named Christophers-MacBook-Pro-christopher-<pid>-orders.…. The v6 test describes behaviour that doesn't change, and passed before and after.
• TestKafkaHosting (Explicit):
  • ShouldHandleEachCommandOnceAcrossTheServicesBuses failed with duplicates.
  • ShouldHandleCommandsSentWhileTheServiceWasDown and ShouldReceiveEventsPublishedWhileABusWithAFixedInstanceIdWasDown failed with timeouts: the messages were missed.
  • ShouldDeliverEveryEventToEveryBus passed before and after.

What was run, locally:

• The whole Obvs.Kafka.Tests suite against Homebrew Kafka on localhost:9092 (with test-topic-3) and the SASL broker from SaslBroker/docker-compose.yml: 106 passed, 0 failed. On their own, the Explicit tests: 19 passed.
• dotnet build Obvs.Kafka for net472, netstandard2.0 and net10.0: no errors or warnings.
• No hosting test waits out the 30-second assignment timeout: a bus that gets no partitions in a shared group still subscribes promptly.

Not tested:

• The fatal and authorization paths are tested only at unit level, with Error values built in the test. I didn't reproduce them against a broker, which would need static membership fencing or ACLs.
• The transient error-handler path isn't reproduced end to end: no test takes a broker down mid-subscription.
• Wiring the host's ILoggerFactory into UseKafka has no test of its own.
• Nothing runs against Event Hubs, Confluent Cloud or MSK themselves.

Test fixture: TestKafkaHosting.ServiceNames now also deletes the run's consumer groups. It skips groups that still have members, because some tests dispose hosts without stopping them.

Follow-ups (not in this PR)

• An exception in Obvs's own payload handling still ends the subscription. That covers an unknown PayloadType, a payload the message deserializer can't read, and a throwing property filter. It's in ConsumeMessage's general catch, not an error from the Kafka client, so I left it alone. It's probably worth the same treatment: skip and log.
• MessageSource.Dispose() is empty, so a host that's disposed without StopAsync keeps its consumers until the process exits. The fixture cleanup above runs into this.
• No core Obvs change was needed.

Sources: Confluent.Kafka 2.16 XML docs (SetErrorHandler, IConsumer.Consume), librdkafka INTRODUCTION.md (fatal errors, unknown or unauthorized topics), librdkafka rdkafka_sasl_oauthbearer.c, KIP-211, Event Hubs quotas.

🤖 Generated with Claude Code