[christopherread]

Next v7 increment: UseKafka() for #84, following UseRabbitMq() (#112) and UseNats() (#118). It also carries message headers, including trace context, over Kafka (#90). Targets v7.

API

builder.Services.AddObvs(obvs => obvs
    .AddService<IOrderMessage>("orders", service => service
        .UseKafka(builder.Configuration.GetSection("Obvs:Orders:Kafka"))   // or .UseKafka(o => o.BootstrapServers = "localhost:9092")
        .SerializedWith(new JsonMessageSerializer(), new JsonMessageDeserializerFactory(typeof(JsonMessageDeserializer<>)))));

• Options: KafkaOptions has BootstrapServers (required). Each service's settings are named options keyed by the service name: IOptionsMonitor<KafkaOptions>.Get("orders"). v6's fetch and batch tuning isn't exposed yet.
• Startup checks: missing brokers or a missing serializer fail at startup, inside the same ObvsConfigurationException as any other problem. The message names the configuration key to set (for example Obvs:Orders:Kafka:BootstrapServers) or the method to call.

How Kafka differs from the other transports

• Topics are created at startup.
  • What: UseKafka creates whichever of the service's four topics don't exist yet: orders.Commands, .Events, .Requests and .Responses. It uses the broker's default partitions and replication, and waits until every partition has a leader.
  • Why: consumers don't create topics, and one that subscribes before its topic exists can take minutes to notice it, missing what's published meanwhile.
  • Existing topics: topics that already exist are left alone, so a cluster that doesn't allow creating topics works as long as they're there.
• Each source has its own consumer group, named from the client ID, the topic and the bus's instance ID. This mirrors RabbitMQ's per-bus queue suffix (v7: UseRabbitMq, options from configuration, per-bus instance ID #112).

Two Kafka transport bugs this exposed and fixes

Both also affect v6. UseKafka avoids the first one; the second fix applies to v6's fluent setup too.

1. Buses in one process shared a consumer group.
   • Cause: v6 names the group after the machine, user and process. Kafka gives each message to only one member of a group, so a second bus in the process missed messages.
   • A second effect: sources for different topics shared that group too. Each new subscription therefore paused all the others for a rebalance, and one consumer that stopped polling stalled every rebalance.
   • Symptoms: ShouldDeliverToEveryBusInTheProcess and ShouldReturnResponsesOverKafka timed out.
   • Fix: UseKafka passes a unique group suffix. v6's fluent setup keeps its group naming.
2. Deadlock when disposing the bus from a subscriber.
   • Cause: messages reached subscribers on the consume thread, and disposing waits for that thread to finish.
   • Fix: messages are now handed to subscribers on the task pool, in order, as in the RabbitMQ fix (RabbitMQ: deliver messages to subscribers off the consumer-dispatch thread #113).

Headers over Kafka

• Sending: the publisher sends ObvsMessageHeaders as Kafka headers, UTF-8 encoded.
• Receiving: the source copies Kafka headers back onto the received message before subscribers see it. The receiver's process span therefore continues the sender's trace.
• Compatibility: v6 receivers ignore Kafka headers, so v6 and v7 services can still exchange messages.

Tests (written first)

• Configuration (no broker): 5 tests in Obvs.Kafka.Tests/Hosting/TestUseKafka.cs. They cover options from configuration and from code, missing brokers (in configuration and in code), and a missing serializer.
• End to end (Explicit, real broker): 8 tests in TestKafkaHosting.cs. They cover:
  • topic creation;
  • commands and events;
  • request and response;
  • every bus in a process receiving every message;
  • two hosts in one process;
  • disposing the bus from a message callback;
  • trace continuation;
  • headers the app sets.
    Service names are unique to each run, so every run creates its topics, and a fixture deletes them afterwards.
• Red first:
  • All 13 tests failed with NotImplementedException before implementation.
  • Once UseKafka worked, the remaining failures matched their causes: the every-bus and request-response tests timed out, the dispose test hung, and the header and trace tests failed on a missing x-tenant header and a different trace ID.
  • Each fix turned its own tests green.
• Mutation check: passing no group suffix makes the every-bus and request-response tests fail again.
• Results:
  • All 10 Kafka broker tests (the 8 new ones plus the 2 existing v6 ones) passed 3 runs in a row against a local Kafka 4.3.1, taking about 40 seconds per run. No topics were left behind.
  • A clean build of the solution has 0 warnings, and all 215 unit tests pass.

Not in this increment

• Use<Transport>() and header support for ActiveMQ and NetMQ.
• Exposing Kafka fetch and batch tuning in KafkaOptions.

🤖 Generated with Claude Code