Kafka: messages published straight after subscribing can be missed #103

In Obvs.Kafka, a message published straight after subscribing to MessageSource.Messages can be silently lost.

Cause

In Obvs.Kafka/MessageSource.cs:

• Subscribe returns too early. Messages blocks on subscribedEvent.Wait() (line 63), but the consumer thread sets that event straight after consumer.Subscribe(_topicName) (lines 83–84). Subscribe only registers interest; partitions are assigned later, during the consumer group join and rebalance, which happens inside Consume.
• New consumers start at the latest offset. The consumer config hard-codes AutoOffsetReset = AutoOffsetReset.Latest (line 47). Each process uses a new consumer group (GroupId = clientId, line 40), so it has no committed offset and starts from the end of the log at the time partitions are assigned.

So anything published between Subscribe returning and partitions being assigned is skipped. On a fresh broker that window is several seconds:

• the default group join delay (group.initial.rebalance.delay.ms = 3s);
• plus creating __consumer_offsets the first time a consumer group joins.

Evidence

Obvs.Kafka.Tests.TestMessageSendAndReceive (subscribe, wait 2s, publish 3 messages) received nothing in CI against a fresh apache/kafka:4.4.0 broker. Publishing succeeded, and the consumer printed no errors. It passed only after #82 added a warm-up step: create the topic, run a throwaway consumer group so __consumer_offsets exists, and set KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS=0. That works around the problem in CI; users of the library still have it.

Possible fixes

• Wait for partitions to be assigned before Subscribe returns, for example by setting subscribedEvent from ConsumerBuilder.SetPartitionsAssignedHandler(...) instead of after consumer.Subscribe. Add a timeout, so a broker with no partitions for the topic doesn't block forever.
• Make the starting offset configurable (for example via KafkaSourceConfiguration) rather than hard-coding Latest. Keep Latest as the default for compatibility.

Done when

• A message published straight after Subscribe returns is received, on a fresh broker.
• TestMessageSendAndReceive passes without needing Run broker-backed transport tests in CI #82's broker warm-up.