BrighterCommand / BrighterCommand/Brighter
Kafka KIP-848 consumer does not surface a missing topic under OnMissingChannel.Assume
- Dominant language
- C#
- Stars
- 2.5k
- Forks
- 296
- Avg merge
- 1d 11h
- Merged PRs (30d)
- 21
Description
### Summary
Under the KIP-848 consumer-group protocol (`GroupProtocol = new ConsumerGroupProtocol()`, added in #4233), `KafkaMessageConsumer` does not surface a missing topic. With `OnMissingChannel.Assume` — which tells Brighter *not* to check whether the infrastructure exists — a receive against a topic that does not exist returns an empty message instead of raising, so the caller silently gets nothing forever.
The classic protocol surfaces it correctly. This is a behavioural divergence between the two protocols, not a configuration problem.
### Observed
Sending to and then receiving from a topic that does not exist, with `OnMissingChannel.Assume` on both publication and subscription, and broker auto-topic-creation disabled:
| Group protocol | Result |
|---|---|
| `ClassicGroupProtocol` | `Paramore.Brighter.ChannelFailureException: Error connecting to Kafka, see inner exception for details`
inner: `ConsumeException: Subscribed topic not available: : Broker: Unknown topic or partition` |
| `ConsumerGroupProtocol` (KIP-848) | **No exception.** `Receive` returns after the timeout with `Header.MessageType == MT_NONE` |
Deterministic — reproduced on every run, both the sync (Reactor) and async (Proactor) paths.
### Reproduction
```
podman compose -f docker-compose-kafka.yaml up -d
dotnet test tests/Paramore.Brighter.Kafka.Tests -f net9.0 \
--filter "FullyQualifiedName~WhenInfrastructureMissingAndAssumeChannel"
```
The generated conformance test `When_infrastructure_missing_and_assume_channel_should_throw_exception` passes for `Kafka / Classic` and `Kafka / PartitionKey` and fails for `Kafka / Consumer`. Note `docker-compose-kafka.yaml` sets `KAFKA_AUTO_CREATE_TOPICS_ENABLE: "false"`, so the topic genuinely does not come into existence.
### Why this is the `Assume` path specifically
`KafkaMessagingGateway.EnsureTopic()` returns immediately for `OnMissingChannel.Assume` and makes no admin call, so nothing checks the topic up front — by design; that is what `Assume` means. The error therefore has to come from the consume itself, and under KIP-848 it does not.
The explicit `OnMissingChannel.Validate` path is unaffected and works on both protocols: it goes through `AdminClient.GetMetadata` in `FindTopic()` and throws `ChannelFailureException: Topic: does not exist` as expected. So this is not a missing broker management API — the same broker image serves both.
Likely the relevant difference is that librdkafka reports the unknown topic in a way the classic path turns into a `ConsumeException` (caught at `KafkaMessageConsumer.Receive` and rethrown as `ChannelFailureException`), while the KIP-848 path neither raises nor latches `_hasFatalError` via `HandleError(Error)`. That last part is a hypothesis — the observed behaviour above is what is confirmed.
### Impact
`OnMissingChannel.Assume` is the "trust me, it's there" option, so a typo'd or undeployed topic is exactly the case where a clear failure matters. On KIP-848 the pump instead reads nothing indefinitely with no error, which is difficult to diagnose in production.
### Current workaround
Not fixed; declared. #4297 adds a generator gate `HasSupportToDetectMissingInfrastructureOnAssume` (default `true`), and `Kafka / Consumer` sets it `false` so only the `assume_channel` conformance test is skipped for that configuration. `validate_channel` is generated and passes. Reverting that flag once this is fixed is the intended end state.
This replaces the coarser workaround from #4233, which set `HasSupportToValidateInfrastructure: false` on `Kafka / Consumer` and so suppressed the `validate_channel` test too — that one works and is now running again.
### Related
- #4233 — the KIP-848 PR; its commit trail records this as "async completion for non-existent topics", and includes a reverted attempt whose message notes that a `RetryableChannel` `UnknownTopicOrPart` retry suppressed the same exception. `RetryableChannel` is not in the tree today.
- #4240 / #4297 — the conformance work that surfaced the divergence.
Contributor guide
Assessment
This issue has not been assessed yet.