net: manager owns net lifecycle (auto-start/close, ownership-gated)

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):

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'],
@@ -114,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:
@@ -187,7 +201,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 +420,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 +487,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)

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