From 8386b4b77abf9f83fb569e158edd4f0688ba4d57 Mon Sep 17 00:00:00 2001 From: Dana Powers Date: Tue, 7 Jul 2026 18:56:09 +0000 Subject: [PATCH 1/2] net: manager owns net lifecycle (auto-start/close, ownership-gated) Make KafkaConnectionManager self-sufficient for direct use: track net ownership (owned iff the manager created it -- net arg is None or a name string, not a passed-in NetBackend instance), auto-start the owned net lazily on the blocking entry points (run/bootstrap), and close it in close(). This lets a direct manager user do KafkaConnectionManager(bootstrap_servers=...) then bootstrap()/run() with no explicit _net.start()/close() -- and advances Phase-D (manager as the entry point, not the compat shim). Ownership gating is what keeps it safe: - Auto-start only on run()/bootstrap() (blocking user entry points), NOT on the internal fire-and-forget call_soon() -- which cluster/sender/coordinator call and which must not spin up a thread. - on_io_thread() guard on the auto-close so close() from a produce callback (on the IO thread) doesn't try to join that thread. - The unit-test harness passes an *instance* (unstarted NetworkSelector it drives via drain()/poll()) -> not owned -> untouched. KafkaNetClient (the legacy poll()-driving shim) manages its net lifecycle explicitly (start/poll/close), so it now resolves + creates its own net and hands the manager a ready instance -- ownership follows creation, and the manager does NOT auto-start/close it (which would collide with the shim's net.poll() driving). Direct KafkaConnectionManager(net=None) users get the auto lifecycle instead. Co-Authored-By: Claude Opus 4.8 (1M context) --- kafka/net/compat.py | 14 +++++++----- kafka/net/manager.py | 26 +++++++++++++++++++++-- test/integration/test_sasl_integration.py | 10 ++++----- 3 files changed, 37 insertions(+), 13 deletions(-) diff --git a/kafka/net/compat.py b/kafka/net/compat.py index 5d32fdad0..2df88018b 100644 --- a/kafka/net/compat.py +++ b/kafka/net/compat.py @@ -4,6 +4,7 @@ import time import kafka.errors as Errors +from kafka.net.backend import resolve_backend from kafka.net.manager import KafkaConnectionManager @@ -21,11 +22,14 @@ def __init__(self, net=None, manager=None, **configs): # _lock is still used by the legacy Coordinator (kafka/coordinator/base.py). # Remove once Coordinator moves to the IO thread (Phase D). self._lock = threading.RLock() - # Backend selection (raw `net`: None | NetBackend | name) is resolved by - # KafkaConnectionManager, not here -- this compat shim is slated for - # removal once callers use the manager directly ("Phase D"). - self._manager = KafkaConnectionManager(net, **configs) if manager is None else manager - self._net = self._manager._net + # This legacy shim owns the net's lifecycle explicitly (start / poll / + # close), so it resolves + creates the net itself and hands the manager + # a ready instance. Passing an instance means the manager does NOT + # auto-start/close it (that would break this shim's poll()-driving) -- + # ownership follows creation. Direct KafkaConnectionManager(net=None) + # users get the manager's auto lifecycle instead. + self._net = resolve_backend(net, configs) + self._manager = KafkaConnectionManager(self._net, **configs) if manager is None else manager @property def cluster(self): diff --git a/kafka/net/manager.py b/kafka/net/manager.py index 76feddf6c..ea15a034d 100644 --- a/kafka/net/manager.py +++ b/kafka/net/manager.py @@ -84,8 +84,16 @@ def __init__(self, net=None, **configs): # `net` is the raw backend selector: a NetBackend instance, a backend # name ('selector'/'asyncio'), or None to auto-detect / default to the - # NetworkSelector. Resolved here (not in the legacy compat shim) so the - # manager remains the durable entry point once compat.py is removed. + # NetworkSelector. Resolving here keeps the manager a self-sufficient + # entry point (durable once compat.py is removed). + # + # Ownership: the manager owns the net's lifecycle iff it created it (a + # name or None), not when a ready NetBackend instance was passed in. + # An owned net is auto-started lazily on first use and closed in + # close(); a passed-in instance is left alone (so the unit-test harness, + # which passes an unstarted NetworkSelector and drives it via + # drain()/poll(), is untouched). + self._owns_net = net is None or isinstance(net, str) self._net = resolve_backend(net, self.config) self.cluster = ClusterMetadata( bootstrap_servers=self.config['bootstrap_servers'], @@ -187,7 +195,14 @@ def bootstrap_async(self, timeout_ms=None, refresh=True): self._bootstrap_future.add_errback(lambda exc: log.error('Bootstrap failed: %s', exc)) return self._bootstrap_future + def _maybe_start(self): + """Start an owned net on first use (idempotent; no-op if already + running or if the net was passed in / is not owned).""" + if self._owns_net: + self._net.start() + def bootstrap(self, timeout_ms=None, refresh=True): + self._maybe_start() self._net.run(self.bootstrap_async, timeout_ms, refresh) @property @@ -399,6 +414,12 @@ def close(self, node_id=None, timeout_ms=None): for conn in list(self._conns.values()): conn.close() self.cluster.close() + # Owned net: the manager created it, so it tears it down too + # (idempotent). A passed-in instance is the caller's to close. + # Skip when called from the IO thread itself (net.close() would + # join that thread) -- same guard the producer's close() uses. + if self._owns_net and not self._net.on_io_thread(): + self._net.close() async def wait_for(self, future, timeout_ms): """Await `future` with a timeout in ms. Raises KafkaTimeoutError on timeout. @@ -460,4 +481,5 @@ def run(self, coro, *args): If no IO thread is running, falls back to driving the loop on the caller thread (legacy behavior). """ + self._maybe_start() return self._net.run(coro, *args) diff --git a/test/integration/test_sasl_integration.py b/test/integration/test_sasl_integration.py index bcfc7500a..504b32ce1 100644 --- a/test/integration/test_sasl_integration.py +++ b/test/integration/test_sasl_integration.py @@ -76,12 +76,11 @@ def test_client(request, sasl_kafka): create_topics(sasl_kafka, [topic_name], num_partitions=1) # Low-level SASL round-trip via KafkaConnectionManager directly (no compat - # shim, no poll()): the started-loop + manager.run(coro) pattern the real - # clients use, so it runs on any net backend (selector or asyncio). + # shim, no poll()): the manager owns + auto-starts/closes its net, so this + # is just bootstrap + manager.run(coro) and runs on any backend. manager = KafkaConnectionManager(**client_params(sasl_kafka, 'client')) - manager._net.start() try: - manager.bootstrap(timeout_ms=5000) + manager.bootstrap(timeout_ms=5000) # auto-starts the owned net async def fetch_metadata(): future = manager.send(MetadataRequest(topics=None, version=1), node_id=None) @@ -90,5 +89,4 @@ async def fetch_metadata(): result = manager.run(fetch_metadata) assert topic_name in [t[1] for t in result.topics] finally: - manager.close() - manager._net.close() + manager.close() # auto-closes the owned net From 817562037200a54752028b2315e762dbbf7b7468 Mon Sep 17 00:00:00 2001 From: Dana Powers Date: Sat, 11 Jul 2026 19:52:39 -0700 Subject: [PATCH 2/2] add context manager --- kafka/net/manager.py | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/kafka/net/manager.py b/kafka/net/manager.py index ea15a034d..ef538e451 100644 --- a/kafka/net/manager.py +++ b/kafka/net/manager.py @@ -122,6 +122,12 @@ def __init__(self, net=None, **configs): self.broker_version_data = BrokerVersionData(self.config['api_version']) self.closed = False + def __enter__(self): + return self + + def __exit__(self, exc_type, exc_val, exc_tb): + self.close() + @property def broker_version(self): if self.broker_version_data is None: