BrighterCommand / BrighterCommand/Brighter

Kafka KIP-848 consumer does not surface a missing topic under OnMissingChannel.Assume

Open
#4,299 0 comments 0 reactions 1 assignee Claimed by @iancooper View on GitHub
.NET 1 - Up Next V10.X
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

Open the contributing guide

Assessment

This issue has not been assessed yet.

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.