diff --git a/kafka/admin/client.py b/kafka/admin/client.py index aff032455..3161fac70 100644 --- a/kafka/admin/client.py +++ b/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): diff --git a/kafka/consumer/group.py b/kafka/consumer/group.py index 63a527a92..84adfbbca 100644 --- a/kafka/consumer/group.py +++ b/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 diff --git a/kafka/net/backend.py b/kafka/net/backend.py index 7f4c57b47..9c77d26fe 100644 --- a/kafka/net/backend.py +++ b/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=''`` 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') diff --git a/kafka/net/compat.py b/kafka/net/compat.py index 8f786e98c..5d32fdad0 100644 --- a/kafka/net/compat.py +++ b/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): diff --git a/kafka/net/manager.py b/kafka/net/manager.py index 1e1d2d44a..aaa40107c 100644 --- a/kafka/net/manager.py +++ b/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'], diff --git a/kafka/producer/kafka.py b/kafka/producer/kafka.py index 2552b8246..abf43479a 100644 --- a/kafka/producer/kafka.py +++ b/kafka/producer/kafka.py @@ -458,6 +458,7 @@ class KafkaProducer: 'proxy_url': None, 'socks5_proxy': None, # deprecated 'kafka_client': KafkaNetClient, + 'net': None, } DEPRECATED_CONFIGS = () diff --git a/test/net/test_backend.py b/test/net/test_backend.py index 2a2d50452..70aa9b6eb 100644 --- a/test/net/test_backend.py +++ b/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() diff --git a/test/net/test_manager.py b/test/net/test_manager.py index fcc45f041..ebf60eac4 100644 --- a/test/net/test_manager.py +++ b/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)