AbsaOSS / AbsaOSS/EventGate

Kafka flush timeout returns 202 with delivery unconfirmed

Aperta
#220 0 commenti 0 reazioni 0 assegnatari Vedi su GitHub
bug
Lingua principale
Python
Stelle
4
Fork
0
Merge medio
1g 5h
PR unite (30g)
8

Descrizione

### 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.

Guida per i contributori

Nessuna guida per i contributori indicizzata per questo repository

Valutazione

Questa issue non è ancora stata valutata.

Ricevi le nuove issue nella tua casella

Un breve riepilogo di issue GitHub adatte ai principianti.