AbsaOSS / AbsaOSS/EventGate

Kafka flush timeout returns 202 with delivery unconfirmed

Abierto
#220 0 comentarios 0 reacciones 0 asignados Ver en GitHub
bug
Lenguaje dominante
Python
Estrellas
4
Forks
0
Merge medio
1 d 5 h
PR fusionados (30 d)
8

Descripción

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

Guía de contribución

No hay ninguna guía de contribución indexada para este repositorio

Evaluación

Este issue todavía no se ha evaluado.

Recibe los nuevos issues en tu correo

Un resumen breve de issues de GitHub para principiantes.