fix: close consumer and producer when initial broker connect fails

[ryan-yuuu]

Summary

Fixes a resource leak in KafkaTable.start() (and the same latent bug in KafkaTableWriter.start()): when the initial broker connect fails — the common "broker not up yet" case — the AIOKafkaConsumer/AIOKafkaProducer that start() created is orphaned. The caller's idiomatic try: await start() ... finally: await stop() cannot help, because stop() reads self._consumer/self._producer, which is still None (the handle is only recorded on success). The orphaned object is garbage-collected unclosed, logging Unclosed AIOKafka* and leaking the socket.

Closes #30.

Root cause

The resource armed aiokafka's __del__ unclosed-resource guard at construction, but the failure-cleanup only became reachable after await consumer.start() / await producer.start():

consumer = AIOKafkaConsumer(...)   # __del__ guard armed here
await consumer.start()             # raises when broker unreachable — orphans the consumer
self._consumer = consumer          # only on success
try: ...
except BaseException:
    await consumer.stop(); ...      # guard starts too late

The violated invariant: a resource's cleanup guard must span the entire window from construction to ownership-transfer. ensure_topic() already did this correctly (its admin.start() is inside a try/finally: close()); the two start() paths did not.

Fix

• KafkaTable.start() → refactored onto a contextlib.AsyncExitStack. Each resource's cleanup (_detach_consumer, then _detach_task) is registered the instant the resource is constructed — before the fallible connect — so any failure or cancellation unwinds them in LIFO order. Full success calls pop_all() to commit ownership to the instance and disarm the stack. An orphaned, never-stopped consumer is now structurally impossible.
• stop() unified with the rollback → teardown extracted into _detach_task() / _detach_consumer(); stop() is now just await self._detach_task(); await self._detach_consumer(), and start()'s stack registers those same callbacks in the matching order — so the two teardown paths can never diverge.
• KafkaTableWriter.start() → the same bug for the producer (not in the original issue report, same root cause), fixed with a proportionate try/except BaseException: await producer.stop(); raise (single resource, single step — an AsyncExitStack there would be overkill).

Behavior is preserved exactly on every path, including the documented "reader died during catch-up counts as started (status == "failed")" contract and the degraded catch-up-timeout path.

Tests

New broker-free, deterministic regression tests in TestStartFailureCleanup (a configurable fake consumer/producer drives every lifecycle path — no GC-timing or __del__-warning dependence):

Test	What it pins
test_start_stops_consumer_when_initial_connect_fails	the issue's exact case — was RED before the fix
test_writer_start_stops_producer_when_initial_connect_fails	parallel producer leak — was RED before the fix
test_start_stops_consumer_when_no_partitions_assigned	post-connect setup failure still cleans up
test_start_cleans_up_when_reader_dies_during_catchup	Mode-2: both resources torn down; started is True, status == "failed"
test_successful_start_commits_consumer_without_stopping_it	pop_all() commit does not over-clean

Written test-first: the two bug tests were confirmed failing for the exact cause before any implementation change.

Verification

• Broker-free unit suite: 197 passed.
• Broker-backed integration suite (Redpanda via testcontainers): 41 passed — covers reader-death, restart, degraded, and all barrier() paths the refactor touches.
• 100% line + branch coverage on ktables/kafka_table.py maintained.
• Real-world check against a dead broker (real aiokafka objects, no mocks): both KafkaTable.start() and KafkaTableWriter.start() now emit no Unclosed AIOKafka* signal.

Note

uv.lock carried a stale ktables version (0.3.0); uv corrected it to 1.1.0 to match pyproject.toml. Unrelated to the fix, but a correct sync — happy to drop it if a minimal diff is preferred.

[github-actions]

Coverage report

Click to see where and how coverage changed

File	Statements	Missing	Coverage	Coverage (new stmts)	Lines missing
ktables
kafka_table.py
Project Total

This report was generated by python-coverage-comment-action