Kafka flush timeout returns 202 with delivery unconfirmed
- Langage dominant
- Python
- Étoiles
- 4
- Forks
- 0
- Merge moyen
- 1 j 5 h
- PR mergées (30 j)
- 8
Description
### Describe the bug
`KafkaWriter.write()` (`src/writers/writer_kafka.py`) retries the flush, and when messages are still pending after the last attempt it logs a `WARNING` and falls through:
```python
if isinstance(remaining, int) and remaining > 0:
logger.warning("Kafka flush timed out with messages still pending.", extra={...})
```
A pure timeout appends nothing to `errors` and raises nothing, so the `if errors:` branch is never reached and `write()` returns normally. `_write_to_all()` counts Kafka in `writers_ok` and the request returns `202`, while the message may never have been delivered.
Only a `WARNING` records it, so an alarm on `level = "ERROR"` is blind to this by construction — and ADR-002's `count(ERROR) == count(5xx)` invariant holds here only because the request is never considered failed at all.
### Steps to Reproduce
1. Configure a topic with a Kafka writer and a reachable broker, so the producer initializes and `produce()` succeeds.
2. Make the broker unable to acknowledge the write while keeping the connection alive — e.g. take the partition leader offline after produce, or set `min.insync.replicas` above the number of live in-sync replicas.
3. `POST /topics/{topic_name}` with a valid message.
4. `flush()` returns `remaining > 0` on all 3 attempts (`KAFKA_FLUSH_RETRIES`, 7s timeout each per `KAFKA_FLUSH_TIMEOUT`) and raises no `KafkaException`.
5. Observe the response: `202`, with `kafka` listed in `writers_ok`.
6. Observe the logs: one `WARNING` ("Kafka flush timed out with messages still pending."), zero `ERROR`.
### Expected state
A message whose delivery was never confirmed must not be reported to the caller as written.
`202` is meant to mean "accepted by every configured sink". Either the response reflects that the Kafka write did not complete, or `writers_ok` stops listing a sink whose delivery is unconfirmed — but a caller must not be told the message landed when the service does not know that it did.
### Impact / Severity
High
### Attachments / Evidence
`src/writers/writer_kafka.py` — flush loop, terminal timeout branch, and the `if errors:` gate that the timeout path never reaches:
```python
# Warn if messages still pending after retries
if isinstance(remaining, int) and remaining > 0:
logger.warning(
"Kafka flush timed out with messages still pending.",
extra={"pending_messages": remaining, "flush_timeout_sec": _KAFKA_FLUSH_TIMEOUT_SEC},
)
duration_ms = round((time.perf_counter() - started_at) * 1000, 2)
if errors: # <- empty on a pure timeout
...
raise WriteError(failure_text)
logger.debug("Kafka accepted the message.", ...) # <- reached instead
```
`errors` is appended to only by `delivery_report` (on a delivery error) and by the two `except KafkaException` blocks. A flush that simply does not drain in time hits none of them.
### Related / References
Two options, and the choice is a product decision rather than a cleanup:
1. **Treat a terminal flush timeout as a `WriteError`.** Correct on the contract, but it turns a current `202` into a `500`, so it is a caller-visible behaviour change.
2. **Keep the `202` and alarm on this specific `WARNING`**, treating unconfirmed delivery as an operational signal rather than a request failure.
Option 1 is the honest one if `202` is meant to mean "accepted by every configured sink". Whichever is chosen, the decision belongs in ADR-002 and `writers_ok` must stop reporting an unconfirmed sink as OK.
Acceptance:
- Decision recorded in ADR-002 §Logging strategy.
- `writers_ok` no longer reports a sink whose delivery was never confirmed.
- Unit test covers flush timeout with `remaining > 0` and no raised exception.
Related: ADR-002 (`adr/002-observability/002-observability.md`), #193, PR #204, #219 — the other invariant gap found in the same review.
Guide de contribution
Aucun guide de contribution indexé pour ce dépôt
Évaluation
Cette issue n'a pas encore été évaluée.