BrighterCommand / BrighterCommand/Brighter
kafka-ci: intermittent "Subscribed topic not available" — generated gateway tests subscribe before topic metadata propagates
- Dominant language
- C#
- Stars
- 2.5k
- Forks
- 296
- Avg merge
- 1d 11h
- Merged PRs (30d)
- 21
Description
`kafka-ci` fails intermittently with `ConsumeException: Subscribed topic not available` on a topic the test itself just created. Seen three times, always the same shape, **never on the same test twice**, and it reproduces on `master` — so it is not any one PR's doing.
## Observations
| where | commit | test | topic |
|---|---|---|---|
| [`master`, job `101816901423`](https://github.com/BrighterCommand/Brighter/actions/runs/34145065842/job/101816901423) | `3bdb635e5` | `Classic.Proactor` | `gen.test.01a07cd0…` |
| [#4323 attempt 1, job `102654267205`](https://github.com/BrighterCommand/Brighter/actions/runs/34406923832/job/102654267205) | `17a9ad98d` | `PartitionKey.Reactor` | `gen.pk.test.01a0881c…` |
| [#4323 attempt 2, job `102666421514`](https://github.com/BrighterCommand/Brighter/actions/runs/34406923832/job/102666421514) | `17a9ad98d` | `Classic.Reactor` | `gen.test.01a08840…` |
Every one is `When_posting_a_message_via_the_messaging_gateway_should_be_received`, and every one is:
```
Paramore.Brighter.ChannelFailureException : Error connecting to Kafka, see inner exception for details
---- Confluent.Kafka.ConsumeException : Subscribed topic not available: gen.test.
```
Reactor *and* Proactor, Classic *and* PartitionKey. Constant failure shape with shifting membership is the signature of an environmental race rather than a defect in any one test.
`kafka-ci` is green on master's three most recent runs (`34399594080`, `34214176968`, `34159782273`), so this is intermittent — a single green run is not evidence it is fixed.
## Mechanism
Each test mints a brand-new topic per run — `KafkaClassicMessageGatewayProvider.GetOrCreateRoutingKey` (`tests/Paramore.Brighter.Kafka.Tests/MessagingGateway/KafkaClassicMessageGatewayProvider.cs:242`) returns `new RoutingKey($"gen.test.{Uuid.New():N}")` — then creates a subscription with `OnMissingChannel.Create`, builds producer and channel, sends, and receives:
```csharp
_publication = provider.CreatePublication(provider.GetOrCreateRoutingKey());
_subscription = provider.CreateSubscription(_publication.Topic!, provider.GetOrCreateChannelName(), OnMissingChannel.Create);
_producer = provider.CreateProducer(_publication);
_channel = provider.CreateChannel(_subscription); // consumer subscribes here
_producer.Send(message);
var received = _channel.Receive(TimeSpan.FromMilliseconds(15000));
```
The consumer subscribes before the broker has finished propagating metadata for the just-created topic, so `_consumer.Consume` raises `ConsumeException`.
**Why it ends the test rather than being ridden out.** `KafkaMessageConsumer.Receive` turns *every* `ConsumeException` into a `ChannelFailureException` (`KafkaMessageConsumer.cs:530-533`), with no transient/fatal distinction — unlike the `KafkaException` catch immediately below it, which does check `Error.IsFatal` (`:536-543`).
In production that is survivable: `Reactor` catches `ChannelFailureException`, sleeps `ChannelFailureDelay` and continues (`src/Paramore.Brighter.ServiceActivator/Reactor.cs:137-144`), as does `Proactor`. The generated gateway tests call `_channel.Receive(...)` **directly, with no pump**, so a condition the product tolerates becomes a red build.
So this is best read as a **test-harness gap, not a product defect** — though the missing transient/fatal split at `:530` is arguably worth its own look, since it denies callers the information the `KafkaException` path gives them.
## Suggested fix
Preferred — close the race at its source: after creating a topic under `OnMissingChannel.Create`, have the test provider wait for metadata to propagate (an admin-client describe/poll loop on the new topic) before the consumer subscribes. This fixes the create-then-subscribe handshake for every generated Kafka gateway test at once.
Alternative, or in addition: have the generated tests retry `Receive` on `ChannelFailureException` the way the pump does, so they exercise the same tolerance the product has.
## Not covered by existing issues
- **#4299** is the *inverse* problem — a missing topic failing to surface under `OnMissingChannel.Assume`.
- **#4281** (drained consumer group one offset short) and **#4190** (KIP-848 support) are unrelated.
This is the Kafka counterpart to **#4324**, which recorded the same constant-shape/shifting-membership pattern for the GCP emulator.
Contributor guide
Assessment
This issue has not been assessed yet.