net: add net= backend selection with auto-detect

kafka/admin/client.py

@@ -210,6 +210,7 @@ class KafkaAdminClient(
         'metrics_num_samples': 2,
         'metrics_sample_window_ms': 30000,
         'kafka_client': KafkaNetClient,
+        'net': None,
     }
 
     def __init__(self, **configs):

kafka/consumer/group.py

@@ -362,6 +362,7 @@ class KafkaConsumer:
         'proxy_url': None,
         'socks5_proxy': None,  # deprecated
         'kafka_client': KafkaNetClient,
+        'net': None,
     }
     DEFAULT_SESSION_TIMEOUT_MS_PRE_KIP_735 = 30000
 

kafka/net/backend.py

@@ -46,6 +46,7 @@
 * **Future factory** -- ``create_future`` (see ``BackendFuture``).
 * **Cross-thread wake** -- ``wakeup``.
 """
+import importlib
 from typing import Any, Callable, Optional, Protocol, Sequence, Tuple, runtime_checkable
 
 
@@ -203,3 +204,83 @@ def create_future(self) -> BackendFuture:
     # --- misc -------------------------------------------------------------
     def wakeup(self) -> None:
         """Interrupt the loop's select() from another thread."""
+
+
+# --- backend selection ----------------------------------------------------
+
+# name -> factory(**config) -> NetBackend. Populated by register_backend();
+# 'selector' is always available, 'asyncio' registers itself in Step 4.
+_BACKENDS = {}
+
+
+def register_backend(name, factory):
+    """Register a named backend factory for ``net='<name>'`` selection."""
+    _BACKENDS[name] = factory
+
+
+def register_backend_lazy(name, module, klass):
+    """Lazy register a factory klass from module. Import is deferred until first use."""
+    def lazy_backend(**configs):
+        backend = getattr(importlib.import_module(module), klass)
+        register_backend(name, backend)
+        return backend(**configs)
+    register_backend(name, lazy_backend)
+
+
+def _detect_async_library():
+    """Best-effort name of the async framework the caller is running under.
+
+    Returns 'asyncio' / 'trio' / ... via sniffio if installed, else 'asyncio'
+    if a running asyncio loop is detected, else None (plain sync context ->
+    caller falls back to the default backend).
+    """
+    try:
+        import sniffio
+        try:
+            return sniffio.current_async_library()
+        except sniffio.AsyncLibraryNotFoundError:
+            pass
+    except ImportError:
+        pass
+    try:
+        import asyncio
+        asyncio.get_running_loop()
+        return 'asyncio'
+    except RuntimeError:
+        return None
+
+
+def resolve_backend(net, config):
+    """Resolve the ``net`` config value to a concrete NetBackend instance.
+
+    Precedence:
+      1. an already-constructed NetBackend instance -> used as-is;
+      2. a string name -> looked up in the registry (raises if unknown);
+      3. None -> auto-detect a running async framework (sniffio / running
+         asyncio loop) and use that backend *if registered*, otherwise fall
+         back to the default ``NetworkSelector``.
+
+    Note (Phase 1): auto-detect only selects the *implementation*; the backend
+    still runs on its own IO thread and the public API still blocks the caller.
+    """
+    if isinstance(net, str):
+        try:
+            factory = _BACKENDS[net]
+        except KeyError:
+            raise ValueError('Unknown net backend %r (available: %s)'
+                             % (net, sorted(_BACKENDS)))
+        return factory(**config)
+    if net is not None:
+        if not isinstance(net, NetBackend):
+            raise TypeError('net must be a NetBackend instance, a backend name, '
+                            'or None; got %r' % (net,))
+        return net
+    # net is None: auto-detect, else default. Auto-detected-but-unregistered
+    # backends fall back silently (an explicit name would have raised above).
+    name = _detect_async_library()
+    if name is None or name not in _BACKENDS:
+        name = 'selector'
+    return _BACKENDS[name](**config)
+
+
+register_backend_lazy('selector', 'kafka.net.selector', 'NetworkSelector')

kafka/net/compat.py

@@ -5,7 +5,6 @@
 
 import kafka.errors as Errors
 from kafka.net.manager import KafkaConnectionManager
-from kafka.net.selector import NetworkSelector
 
 
 log = logging.getLogger(__name__)
@@ -22,8 +21,11 @@ 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()
-        self._net = NetworkSelector(**configs) if net is None else net
-        self._manager = KafkaConnectionManager(self._net, **configs) if manager is None else manager
+        # 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
 
     @property
     def cluster(self):

kafka/net/manager.py

@@ -7,6 +7,7 @@
 
 from .connection import KafkaConnection
 from .metrics import KafkaManagerMetrics
+from kafka.net.backend import resolve_backend
 from kafka.cluster import ClusterMetadata
 import kafka.errors as Errors
 from kafka.net.transport import KafkaSSLTransport
@@ -59,7 +60,7 @@ class KafkaConnectionManager:
     }
     _VALID_DNS_LOOKUP_MODES = ('use_all_dns_ips', 'resolve_canonical_bootstrap_servers_only')
 
-    def __init__(self, net, **configs):
+    def __init__(self, net=None, **configs):
         self.config = copy.copy(self.DEFAULT_CONFIG)
         for key in self.config:
             if key in configs:
@@ -75,7 +76,11 @@ def __init__(self, net, **configs):
                 log.warning('socks5_proxy is deprecated, use proxy_url instead')
                 self.config['proxy_url'] = configs['socks5_proxy']
 
-        self._net = net
+        # `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.
+        self._net = resolve_backend(net, configs)
         self.cluster = ClusterMetadata(
             bootstrap_servers=self.config['bootstrap_servers'],
             metadata_max_age_ms=self.config['metadata_max_age_ms'],

kafka/producer/kafka.py

@@ -458,6 +458,7 @@ class KafkaProducer:
         'proxy_url': None,
         'socks5_proxy': None,  # deprecated
         'kafka_client': KafkaNetClient,
+        'net': None,
     }
 
     DEPRECATED_CONFIGS = ()

test/net/test_backend.py

@@ -5,9 +5,14 @@
 ``on_io_thread()`` behaves correctly. Step 4's AsyncioBackend will be held to
 the same isinstance/method-presence checks.
 """
+import asyncio
 import threading
 
-from kafka.net.backend import NetBackend, Transport
+import pytest
+
+from kafka.net.backend import (
+    NetBackend, Transport, resolve_backend, register_backend, _BACKENDS,
+)
 from kafka.net.selector import NetworkSelector
 from kafka.net.transport import KafkaTCPTransport
 
@@ -82,3 +87,100 @@ async def where():
             assert threading.current_thread() is not net._io_thread
         finally:
             net.close()
+
+
+@pytest.fixture
+def clean_registry():
+    """Snapshot/restore the backend registry so register_backend() in a test
+    doesn't leak into others."""
+    saved = dict(_BACKENDS)
+    try:
+        yield
+    finally:
+        _BACKENDS.clear()
+        _BACKENDS.update(saved)
+
+
+class TestResolveBackend:
+    def test_none_defaults_to_networkselector(self):
+        b = resolve_backend(None, {})
+        assert isinstance(b, NetworkSelector)
+        assert isinstance(b, NetBackend)
+
+    def test_explicit_instance_used_as_is(self):
+        sel = NetworkSelector()
+        assert resolve_backend(sel, {}) is sel
+
+    def test_name_selector_resolves(self):
+        assert isinstance(resolve_backend('selector', {}), NetworkSelector)
+
+    def test_unknown_name_raises(self):
+        with pytest.raises(ValueError, match='Unknown net backend'):
+            resolve_backend('bogus', {})
+
+    def test_asyncio_name_unregistered_raises(self):
+        # In Phase-1/Step-3 the asyncio backend is not registered yet; an
+        # explicit request for it is a hard error (an auto-detect is not).
+        with pytest.raises(ValueError, match='Unknown net backend'):
+            resolve_backend('asyncio', {})
+
+    def test_non_backend_instance_raises(self):
+        with pytest.raises(TypeError):
+            resolve_backend(object(), {})
+
+    def test_config_passed_through_to_default(self):
+        b = resolve_backend(None, {'client_id': 'resolver-test'})
+        assert b.config['client_id'] == 'resolver-test'
+
+    def test_name_resolution_passes_config(self, clean_registry):
+        seen = {}
+
+        def factory(**config):
+            seen.update(config)
+            return NetworkSelector(**config)
+
+        register_backend('dummy', factory)
+        resolve_backend('dummy', {'client_id': 'via-name'})
+        assert seen.get('client_id') == 'via-name'
+
+    def test_autodetect_uses_registered_backend_in_loop(self, clean_registry):
+        sentinel = NetworkSelector()
+        register_backend('asyncio', lambda **cfg: sentinel)
+
+        async def main():
+            return resolve_backend(None, {})
+
+        assert asyncio.run(main()) is sentinel
+
+    def test_autodetect_falls_back_when_unregistered_in_loop(self, clean_registry):
+        _BACKENDS.pop('asyncio', None)  # ensure not registered
+
+        async def main():
+            return resolve_backend(None, {})
+
+        assert isinstance(asyncio.run(main()), NetworkSelector)
+
+    def test_no_running_loop_defaults_to_selector(self):
+        assert isinstance(resolve_backend(None, {}), NetworkSelector)
+
+
+class TestClientNetConfig:
+    def test_default_config_has_net_none(self):
+        from kafka.producer.kafka import KafkaProducer
+        from kafka.consumer.group import KafkaConsumer
+        from kafka.admin.client import KafkaAdminClient
+        for cls in (KafkaProducer, KafkaConsumer, KafkaAdminClient):
+            assert cls.DEFAULT_CONFIG['net'] is None, cls.__name__
+
+    def test_kafkanetclient_resolves_net(self):
+        from kafka.net.compat import KafkaNetClient
+        c = KafkaNetClient(bootstrap_servers='localhost:9092')
+        assert isinstance(c._net, NetworkSelector)
+        c._net.close()
+
+    def test_kafkanetclient_honors_explicit_instance(self):
+        from kafka.net.compat import KafkaNetClient
+        sel = NetworkSelector()
+        c = KafkaNetClient(net=sel, bootstrap_servers='localhost:9092')
+        assert c._net is sel
+        sel.close()

test/net/test_manager.py

@@ -19,6 +19,20 @@ def net():
     return NetworkSelector()
 
 
+class TestKafkaConnectionManagerNetResolution:
+    """Backend selection lives on the manager (not the compat shim)."""
+
+    def test_default_resolves_to_selector(self):
+        m = KafkaConnectionManager()
+        assert isinstance(m._net, NetworkSelector)
+
+    def test_explicit_instance_used_as_is(self, net):
+        assert KafkaConnectionManager(net)._net is net
+
+    def test_name_resolves(self):
+        assert isinstance(KafkaConnectionManager('selector')._net, NetworkSelector)
+
+
 class TestKafkaConnectionManagerConfig:
     def test_default_config(self, net):
         m = KafkaConnectionManager(net)