KafkaTable reader task dies permanently on broker disconnect (never reconnects/recovers)

[ryan-yuuu]

Summary

A KafkaTable's reader task dies permanently when the broker connection is lost in a way that raises out of the fetch loop. Once the reader task exits with an exception, the table is latched into a terminal failed / frozen state for the lifetime of the object — it never reconnects or recovers, even after the broker comes back and the topic is fully healthy. The only remedy is to construct a brand-new table.

Mechanism

The reader loop KafkaTable._run (kafka_table.py) is:

while True:
    batches = await consumer.getmany(timeout_ms=self._poll_timeout_ms)
    ...
    # (position()/end_offsets() calls for the catch-up gate and barriers)

There is no reconnect/retry wrapper around the loop. The only in-loop exception that is caught is IllegalStateError from position() (a transient unassignment). Any other exception escaping getmany() / position() / end_offsets() — i.e. the fatal aiokafka errors a broker disconnect can surface — propagates out of the task.

_on_reader_done (the task's done-callback) then latches the failure permanently:

def _on_reader_done(self, task):
    if task.cancelled():
        return
    exc = task.exception()
    if exc is not None:
        self._failure = exc
        self._failed.set()
        logger.error("KafkaTable reader for topic=%s DIED (...); table is FROZEN at last applied state (... status=failed)", ...)

After this:

• failure stays set forever (there is no code path that clears it or re-spawns the reader task);
• reads serve the frozen last-applied snapshot;
• wait_until_caught_up() / barrier() return False (they short-circuit on self._failure);
• status is permanently "failed".

Nothing restarts the reader when the broker returns, so a transient outage becomes a permanent death.

The in-loop comment states the intent — "transient outages don't raise — getmany returns empty" — but in practice a broker disconnect (broker restart, network partition, host sleep/resume) can raise a fatal error out of the fetch/position calls, and that path is terminal.

Reproduction

1. Start a broker and create a topic.
2. Construct a KafkaTable over that topic and await table.start() — let it catch up (is_caught_up / status == "caught_up").
3. Sever the broker connection in a way that raises out of the loop, not a clean idle: kill/restart the broker process, or drop the network (e.g. a laptop sleep/resume), while the reader is live.
4. Observe the reader task exit with an exception and _on_reader_done log ... DIED ...; table is FROZEN ... status=failed.
5. Bring the broker back and let the topic become fully reachable again.

Also observed in the wild: a long-lived reader left running for hours across network-loss / sleep cycles reaches the same terminal state.

Failing case (observed)

• After step 5, with the broker and topic healthy again, the table stays dead: failure is still set, status == "failed", wait_until_caught_up() / barrier() keep returning False, and reads serve stale (frozen) data indefinitely.
• A freshly constructed KafkaTable over the same topic and broker catches up and reads correctly — proving the broker, topic, and data are healthy and the fault is confined to the dead reader object.
• No reconnection or self-recovery ever occurs for the lifetime of the process holding that table.

Downstream impact: consumers that cache a single long-lived reader inherit this permanent death. For example, calfkit's client.mesh (ControlPlaneView over a KafkaTable) surfaces it as a permanent MeshUnavailableError(reason="reader_dead") that only a process restart clears.

Environment

• ktables 1.1.1
• aiokafka 0.13.0
• Python 3.12
• First observed 2026-07-03.

[ryan-yuuu]

Closing: investigation refuted this issue's core mechanism.

A broker disconnect does not kill the reader — getmany() returns empty and aiokafka retries transport failures. Verified empirically against a real broker on a confirmed-silent port (with the port proven REFUSED during the outage and cluster metadata asserted single-broker) and traced in aiokafka 0.14.0 source (transport-level fetch failures are retried in _proc_fetch_request; getmany() only raises for a broker-returned non-retriable per-partition code). A kill/pause of the broker left the caught-up table alive and it recovered.

The permanent-death fragility was real, but its triggers were two other, adjacent paths — both fixed in #34:

• a raising on_set/on_delete apply-observer callback (the most likely cause of the wild reader_dead, since calfkit's ControlPlaneView supplies these hooks) — now logged with the record key + traceback and counted via stats.callback_errors, never fatal;
• a transient IllegalStateError from the catch-up position() under a group_id=None metadata blip — now tolerated, mirroring the existing barrier() guard (see ADR 0003).

Not fixed here, tracked in #35: general recovery from any reader death, including genuinely non-retriable errors like TopicAuthorizationFailedError (terminal by design). That's the real remaining gap and a different, larger change (a reader supervisor / restart()).

If a reader_dead recurs after #34, please reopen with the actual failure exception type — that pins whether it was the callback path (fixed) or the supervisor gap (#35).