From 8cd77abde4ce75e2b4ff9580d3a44a7670cdb5e8 Mon Sep 17 00:00:00 2001 From: Dana Powers Date: Wed, 1 Jul 2026 13:43:31 +0000 Subject: [PATCH 1/5] net: define the NetBackend loop-surface contract Add kafka/net/backend.py: NetBackend, a runtime_checkable Protocol capturing the event-loop surface the rest of kafka-python depends on through _net -- lifecycle (start/stop/close/on_io_thread), scheduling (call_soon*/call_at/ call_later/cancel), awaitable primitives (sleep/wait_read/wait_write), the cross-thread run() bridge, create_future(), and wakeup/unregister_event. NetworkSelector is the reference implementation and conforms structurally; an asyncio (and later Twisted) backend will implement the same surface so it can be swapped in via net= without touching core code. Intentionally excluded and documented as such: the legacy single-tick poll(timeout_ms) driver (only the selector-bound KafkaNetClient compat shim uses it; asyncio has no bounded equivalent), and register_event/selector internals. Add on_io_thread() to NetworkSelector as the clean form of the "current_thread() is _io_thread" check, and route the one external caller (KafkaProducer.close) through it instead of reaching into the private attribute -- behavior-identical, and answerable by alternate backends. Add test/net/test_backend.py: isinstance/method-presence conformance for NetworkSelector and on_io_thread() behavior (True on the loop, False off it). No behavior change; full unit suite green. Co-Authored-By: Claude Opus 4.8 (1M context) --- kafka/net/backend.py | 109 +++++++++++++++++++++++++++++++++++++-- kafka/net/selector.py | 10 ++++ kafka/producer/kafka.py | 3 +- test/net/test_backend.py | 65 +++++++++++++++++++++++ 4 files changed, 182 insertions(+), 5 deletions(-) create mode 100644 test/net/test_backend.py diff --git a/kafka/net/backend.py b/kafka/net/backend.py index 58cba7f2c..6593ef588 100644 --- a/kafka/net/backend.py +++ b/kafka/net/backend.py @@ -1,11 +1,44 @@ -"""Pluggable async-backend contracts. +"""The pluggable async-backend contract. -For now this holds the :class:`BackendFuture` contract -- the surface of the +``NetBackend`` is the interface the rest of kafka-python depends on for its +event loop: the surface that ``KafkaProducer`` / ``KafkaConsumer`` / +``KafkaAdminClient`` (and the manager, cluster, connection, coordinator, +fetcher, sender) reach for through ``self._net`` / ``manager._net``. + +``NetworkSelector`` (``kafka/net/selector.py``) is the reference +implementation; an asyncio backend (and eventually Twisted) implements the +same surface so it can be swapped in via ``net=`` without touching core code. + +The :class:`BackendFuture` contract is the surface of the loop-awaitable futures a backend hands out from ``net.create_future()``. The selector's implementation is ``kafka.net.selector.SelectorFuture``; an asyncio (and eventually Twisted) backend supplies its own. + +Two things are intentionally **not** part of the contract: + +* ``poll(timeout_ms, future=...)`` -- the legacy single-tick driver. Its only + remaining caller is the ``KafkaNetClient`` compat shim + (``kafka/net/compat.py``), which is selector-bound; asyncio has no bounded + single-tick equivalent (``run_forever`` never returns, ``run_until_complete`` + runs to a specific future). New code does not use it. +* ``register_event`` / selector internals -- backend-private plumbing. + ``unregister_event`` is included only because the transport calls it on the + cleanup path. + +Method families: + +* **Lifecycle** -- ``start`` / ``stop`` / ``close`` / ``on_io_thread``. +* **Scheduling** -- ``call_soon`` / ``call_soon_threadsafe`` / + ``call_soon_with_future`` / ``call_at`` / ``call_later`` / ``cancel``. +* **Awaitable primitives** -- ``sleep`` / ``wait_read`` / ``wait_write``. + These return a backend-specific awaitable (a ``KernelEvent`` for the + selector; an ``async def`` result for asyncio) -- core coroutines only + ``await`` them. +* **Cross-thread bridge** -- ``run`` (schedule on the loop, block the caller). +* **Future factory** -- ``create_future`` (see ``BackendFuture``). +* **Cross-thread wake** -- ``wakeup``. """ -from typing import Any, Callable, Protocol, runtime_checkable +from typing import Any, Callable, Optional, Protocol, runtime_checkable @runtime_checkable @@ -59,3 +92,73 @@ def add_both(self, f: Callable, *args: Any, **kwargs: Any) -> 'BackendFuture': . def chain(self, future: 'BackendFuture') -> 'BackendFuture': ... def succeeded(self) -> bool: ... def failed(self) -> bool: ... + + +@runtime_checkable +class NetBackend(Protocol): + """Structural contract for a pluggable async event-loop backend. + + ``runtime_checkable`` so conformance can be asserted with ``isinstance``; + note that only checks member *presence*, not signatures. ``NetworkSelector`` + satisfies this structurally (no explicit inheritance needed). + """ + + # --- lifecycle -------------------------------------------------------- + def start(self) -> None: + """Spawn/attach the IO thread that runs the loop. Idempotent.""" + + def stop(self, timeout_ms: Optional[float] = None) -> None: + """Stop the loop and join the IO thread. Idempotent.""" + + def close(self) -> None: + """Stop (if running) and release loop resources. Idempotent.""" + + def on_io_thread(self) -> bool: + """True if the caller is running on this backend's IO thread.""" + + # --- scheduling ------------------------------------------------------- + def call_soon(self, task: Any) -> Any: + """Enqueue a coroutine/callable to run on the next loop iteration.""" + + def call_soon_threadsafe(self, callback: Any) -> Any: + """``call_soon`` from another thread; wakes the loop.""" + + def call_soon_with_future(self, coro: Any, *args: Any) -> BackendFuture: + """Schedule ``coro`` and return a future that resolves with its result.""" + + def call_at(self, when: float, task: Any) -> Any: + """Schedule ``task`` to run at absolute monotonic time ``when``.""" + + def call_later(self, delay: float, task: Any) -> Any: + """Schedule ``task`` to run after ``delay`` seconds.""" + + def cancel(self, task: Any) -> None: + """Cancel a scheduled task/timer previously returned by call_*.""" + + # --- awaitable primitives (core coroutines only await these) ---------- + def sleep(self, delay: float) -> Any: + """Awaitable that resolves after ``delay`` seconds.""" + + def wait_read(self, fileobj: Any, timeout_at: Optional[float] = None) -> Any: + """Awaitable that resolves when ``fileobj`` is readable.""" + + def wait_write(self, fileobj: Any, timeout_at: Optional[float] = None) -> Any: + """Awaitable that resolves when ``fileobj`` is writable.""" + + # --- cross-thread bridge --------------------------------------------- + def run(self, coro: Any, *args: Any) -> Any: + """Schedule ``coro`` on the loop, block the calling thread, return/raise. + + Raises ``RuntimeError`` if called from the IO thread itself. + """ + + # --- future factory --------------------------------------------------- + def create_future(self) -> BackendFuture: + """Create a loop-awaitable future (see ``BackendFuture``).""" + + # --- misc ------------------------------------------------------------- + def wakeup(self) -> None: + """Interrupt the loop's select() from another thread.""" + + def unregister_event(self, fileobj: Any, event: Any) -> None: + """Drop a registered read/write interest (transport cleanup path).""" diff --git a/kafka/net/selector.py b/kafka/net/selector.py index 7726a37b9..41fb15429 100644 --- a/kafka/net/selector.py +++ b/kafka/net/selector.py @@ -277,6 +277,16 @@ def start(self): self._io_thread = t t.start() + def on_io_thread(self): + """True if the caller is running on this backend's IO thread. + + The clean form of the ``current_thread() is _io_thread`` identity + check; callers use it to avoid blocking the loop on itself (e.g. a + producer ``close()`` invoked from a produce callback). Part of the + NetBackend contract so alternate backends can answer it their own way. + """ + return self._io_thread is not None and threading.current_thread() is self._io_thread + def stop(self, timeout_ms=None): """Signal run_forever() to exit and join the IO thread. diff --git a/kafka/producer/kafka.py b/kafka/producer/kafka.py index 20df36f00..2552b8246 100644 --- a/kafka/producer/kafka.py +++ b/kafka/producer/kafka.py @@ -695,8 +695,7 @@ def __getattr__(self, name): log.info("%s: Closing the Kafka producer with %s secs timeout.", str(self), timeout) self.flush(timeout) - on_io_thread = bool(self._net._io_thread is not None - and threading.current_thread() is self._net._io_thread) + on_io_thread = self._net.on_io_thread() if timeout > 0: if on_io_thread: log.warning("%s: Overriding close timeout %s secs to 0 in order to" diff --git a/test/net/test_backend.py b/test/net/test_backend.py new file mode 100644 index 000000000..f75df021a --- /dev/null +++ b/test/net/test_backend.py @@ -0,0 +1,65 @@ +"""Conformance tests for the NetBackend contract (kafka/net/backend.py). + +NetworkSelector is the reference implementation; these pin that it satisfies +the NetBackend Protocol structurally and that the shared lifecycle helper +``on_io_thread()`` behaves correctly. Step 4's AsyncioBackend will be held to +the same isinstance/method-presence checks. +""" +import threading + +from kafka.net.backend import NetBackend +from kafka.net.selector import NetworkSelector + + +# The full contract surface, kept here so a missing/renamed method fails loudly. +CONTRACT_METHODS = ( + 'start', 'stop', 'close', 'on_io_thread', + 'call_soon', 'call_soon_threadsafe', 'call_soon_with_future', + 'call_at', 'call_later', 'cancel', + 'sleep', 'wait_read', 'wait_write', + 'run', 'create_future', 'wakeup', 'unregister_event', +) + + +class TestNetBackendContract: + def test_networkselector_satisfies_protocol(self): + assert isinstance(NetworkSelector(), NetBackend) + + def test_plain_object_is_not_netbackend(self): + assert not isinstance(object(), NetBackend) + + def test_partial_impl_is_not_netbackend(self): + class Partial: + def start(self): + pass + # missing everything else + assert not isinstance(Partial(), NetBackend) + + def test_all_contract_methods_present_and_callable(self): + net = NetworkSelector() + for name in CONTRACT_METHODS: + assert callable(getattr(net, name)), name + + def test_poll_is_not_part_of_contract(self): + # poll() exists on NetworkSelector (legacy) but is intentionally not + # in the NetBackend surface; asyncio has no equivalent. + assert 'poll' not in CONTRACT_METHODS + + +class TestOnIoThread: + def test_false_before_start(self): + net = NetworkSelector() + assert net.on_io_thread() is False + + def test_false_from_other_thread_true_on_loop(self): + net = NetworkSelector() + net.start() + try: + async def where(): + return net.on_io_thread() + # Runs on the IO thread -> True; the calling test thread -> False. + assert net.run(where) is True + assert net.on_io_thread() is False + assert threading.current_thread() is not net._io_thread + finally: + net.close() From f0d344592f7051d8ceef425868d3a7924f2aa073 Mon Sep 17 00:00:00 2001 From: Dana Powers Date: Wed, 1 Jul 2026 14:35:32 +0000 Subject: [PATCH 2/5] net: replace fd-readiness with a create_connection seam asyncio and Twisted own the socket (DNS/connect/TLS/buffering) and don't expose portable fd-readiness (asyncio's Proactor loop has no add_reader). The Step 2 NetBackend leaked the selector's readiness model (wait_read/wait_write/ unregister_event), which doesn't port. Raise the seam to connection creation. The split already exists and already matches: KafkaConnection implements the asyncio.Protocol callback surface and KafkaTCPTransport implements both asyncio.Transport and Twisted ITransport method names. - NetBackend: drop wait_read/wait_write/unregister_event (selector-private, zero core callers); add async create_connection(protocol, host, port, *, ssl, ssl_check_hostname, proxy_url, socket_options, timeout_at) -> Transport, and a small Transport Protocol (the subset KafkaConnection drives). - NetworkSelector.create_connection: move manager._build_transport's body behind the seam (inet.create_connection + KafkaTCPTransport/KafkaSSLTransport + handshake). It returns the transport; the caller wires connection_made() after its "closed during connect" check (behavior-preserving). wait_read/ wait_write/unregister_event remain as private selector methods (still used by transport.py/inet.py). proxy_url still flows to KafkaNetSocket unchanged (selector-only proxy). - manager._connect calls self._net.create_connection(...); _build_transport and the now-unused inet/transport imports are removed. Tests: update the MockBroker harness (both attach() variants) to patch net.create_connection instead of the removed manager._build_transport, and rewrite the two direct-patch manager tests against the new seam. Conformance test drops the readiness prims, adds create_connection + a Transport check. No behavior change; full unit suite green. Co-Authored-By: Claude Opus 4.8 (1M context) --- kafka/net/backend.py | 82 ++++++++++++++++++++++++++++++---------- kafka/net/manager.py | 30 ++++----------- kafka/net/selector.py | 28 ++++++++++++++ test/mock_broker.py | 20 +++++----- test/net/test_backend.py | 33 ++++++++++++---- test/net/test_manager.py | 40 ++++++++++---------- 6 files changed, 153 insertions(+), 80 deletions(-) diff --git a/kafka/net/backend.py b/kafka/net/backend.py index 6593ef588..630eed2f9 100644 --- a/kafka/net/backend.py +++ b/kafka/net/backend.py @@ -14,31 +14,39 @@ selector's implementation is ``kafka.net.selector.SelectorFuture``; an asyncio (and eventually Twisted) backend supplies its own. -Two things are intentionally **not** part of the contract: - +Networking is a **connection seam**, not fd-readiness. asyncio and Twisted own +the socket (DNS, connect, TLS, buffering) and drive protocol callbacks; they do +not expose portable fd-readiness (asyncio's Proactor loop has no ``add_reader``, +Twisted never exposes arbitrary-fd readiness). So the backend provides +``create_connection`` -- given a ``KafkaConnection`` (which already implements +the ``asyncio.Protocol`` surface) and an endpoint, it establishes the transport +and wires the two together. The returned object satisfies the small +:class:`Transport` protocol (the subset ``KafkaConnection`` drives). + +Three things are intentionally **not** part of the contract: + +* ``wait_read`` / ``wait_write`` / ``unregister_event`` -- the low-level + fd-readiness primitives. They are the *selector's* private mechanism (used + only inside ``kafka/net/transport.py`` + ``inet.py``, zero core callers) and + do not port to asyncio/Twisted. The connection seam replaces them. * ``poll(timeout_ms, future=...)`` -- the legacy single-tick driver. Its only remaining caller is the ``KafkaNetClient`` compat shim (``kafka/net/compat.py``), which is selector-bound; asyncio has no bounded - single-tick equivalent (``run_forever`` never returns, ``run_until_complete`` - runs to a specific future). New code does not use it. + single-tick equivalent. * ``register_event`` / selector internals -- backend-private plumbing. - ``unregister_event`` is included only because the transport calls it on the - cleanup path. Method families: * **Lifecycle** -- ``start`` / ``stop`` / ``close`` / ``on_io_thread``. * **Scheduling** -- ``call_soon`` / ``call_soon_threadsafe`` / ``call_soon_with_future`` / ``call_at`` / ``call_later`` / ``cancel``. -* **Awaitable primitives** -- ``sleep`` / ``wait_read`` / ``wait_write``. - These return a backend-specific awaitable (a ``KernelEvent`` for the - selector; an ``async def`` result for asyncio) -- core coroutines only - ``await`` them. +* **Timing** -- ``sleep`` (backend-specific awaitable; core coroutines await it). +* **Connection** -- ``create_connection`` (returns a :class:`Transport`). * **Cross-thread bridge** -- ``run`` (schedule on the loop, block the caller). * **Future factory** -- ``create_future`` (see ``BackendFuture``). * **Cross-thread wake** -- ``wakeup``. """ -from typing import Any, Callable, Optional, Protocol, runtime_checkable +from typing import Any, Optional, Protocol, Sequence, Tuple, runtime_checkable @runtime_checkable @@ -94,6 +102,24 @@ def succeeded(self) -> bool: ... def failed(self) -> bool: ... +@runtime_checkable +class Transport(Protocol): + """The transport surface a backend's ``create_connection`` returns. + + The subset of the ``asyncio.Transport`` / Twisted ``ITransport`` surface + that ``KafkaConnection`` actually drives. The selector's + ``KafkaTCPTransport`` and an asyncio-transport adapter both satisfy it. + """ + + def write(self, data: bytes) -> None: ... + def close(self) -> None: ... + def abort(self, error: Any = None) -> None: ... + def is_closing(self) -> bool: ... + def pause_reading(self) -> None: ... + def resume_reading(self) -> None: ... + def host_port(self) -> Tuple[str, int]: ... + + @runtime_checkable class NetBackend(Protocol): """Structural contract for a pluggable async event-loop backend. @@ -135,15 +161,34 @@ def call_later(self, delay: float, task: Any) -> Any: def cancel(self, task: Any) -> None: """Cancel a scheduled task/timer previously returned by call_*.""" - # --- awaitable primitives (core coroutines only await these) ---------- + # --- timing (core coroutines await this) ------------------------------ def sleep(self, delay: float) -> Any: """Awaitable that resolves after ``delay`` seconds.""" - def wait_read(self, fileobj: Any, timeout_at: Optional[float] = None) -> Any: - """Awaitable that resolves when ``fileobj`` is readable.""" - - def wait_write(self, fileobj: Any, timeout_at: Optional[float] = None) -> Any: - """Awaitable that resolves when ``fileobj`` is writable.""" + # --- connection seam -------------------------------------------------- + async def create_connection( + self, + protocol: Any, + host: str, + port: int, + *, + ssl: Any = None, + ssl_check_hostname: bool = True, + proxy_url: Optional[str] = None, + socket_options: Sequence[Any] = (), + timeout_at: Optional[float] = None, + ) -> Transport: + """Establish and return a connected :class:`Transport` to ``host:port``. + + The backend owns DNS, connect, TLS and (where supported) proxying. The + *caller* wires the transport to the protocol afterwards via + ``protocol.connection_made(transport)`` (so manager-level policy such as + the "closed during connect" check runs first). ``protocol`` (a + ``KafkaConnection``) is passed because backends that own the socket + (asyncio/Twisted) need it at connect time to receive transport events; + they buffer inbound data until ``connection_made`` is called. Backends + without native proxy support raise when ``proxy_url`` is set. + """ # --- cross-thread bridge --------------------------------------------- def run(self, coro: Any, *args: Any) -> Any: @@ -159,6 +204,3 @@ def create_future(self) -> BackendFuture: # --- misc ------------------------------------------------------------- def wakeup(self) -> None: """Interrupt the loop's select() from another thread.""" - - def unregister_event(self, fileobj: Any, event: Any) -> None: - """Drop a registered read/write interest (transport cleanup path).""" diff --git a/kafka/net/manager.py b/kafka/net/manager.py index 55501a2ea..220cef7cf 100644 --- a/kafka/net/manager.py +++ b/kafka/net/manager.py @@ -5,10 +5,8 @@ import socket import time -from .inet import create_connection from .connection import KafkaConnection from .metrics import KafkaManagerMetrics -from .transport import KafkaSSLTransport, KafkaTCPTransport from kafka.cluster import ClusterMetadata import kafka.errors as Errors from kafka.net.wakeup_notifier import WakeupNotifier @@ -203,33 +201,19 @@ def close_idle_connections(self): def ssl_enabled(self): return self.config['security_protocol'] in ('SSL', 'SASL_SSL') - async def _build_transport(self, node, timeout_at=None): - sock = await create_connection(self._net, node.host, node.port, - self.config['socket_options'], - proxy_url=self.config['proxy_url'], - timeout_at=timeout_at) - if self.ssl_enabled: - ssl_configs = {key: value - for key, value in self.config.items() - if key.startswith('ssl_')} - transport = KafkaSSLTransport(self._net, sock, host=node.host, **ssl_configs) - else: - transport = KafkaTCPTransport(self._net, sock, host=node.host) - - try: - await transport.handshake() - except Exception as e: - raise Errors.KafkaConnectionError('Handshake failed: %s' % e) - else: - return transport - async def _connect(self, node, conn, reset_backoff_on_connect=True, timeout_at=None): # Tracks ownership of the freshly built transport: while non-None it is # ours to clean up (the connection hasn't taken it over yet), so the # finally clause closes it. Cleared once connection_made() succeeds. transport = None try: - transport = await self._build_transport(node, timeout_at=timeout_at) + transport = await self._net.create_connection( + conn, node.host, node.port, + ssl=self._build_ssl_context() if self.ssl_enabled else None, + ssl_check_hostname=self.config['ssl_check_hostname'], + proxy_url=self.config['proxy_url'], + socket_options=self.config['socket_options'], + timeout_at=timeout_at) # The connection (or the whole manager) may have been closed while # we were building the transport. Handing it to connection_made() # would flip the conn back to `initializing` and resurrect a diff --git a/kafka/net/selector.py b/kafka/net/selector.py index 41fb15429..55cf685a6 100644 --- a/kafka/net/selector.py +++ b/kafka/net/selector.py @@ -11,6 +11,8 @@ import kafka.errors as Errors from kafka.future import Future +from kafka.net.inet import create_connection as _inet_create_connection +from kafka.net.transport import KafkaSSLTransport, KafkaTCPTransport from kafka.version import __version__ @@ -485,6 +487,32 @@ def create_future(self): """ return SelectorFuture() + async def create_connection(self, protocol, host, port, *, ssl=None, + ssl_check_hostname=True, proxy_url=None, + socket_options=(), timeout_at=None): + """Establish and return a connected transport to host:port. + + The selector owns the raw socket: DNS + non-blocking connect (with + optional SOCKS5/HTTP-CONNECT proxy via KafkaNetSocket), then wraps it + in a TCP or SSL transport and runs the TLS handshake. ``protocol`` (the + KafkaConnection) is not used here -- the caller wires it via + ``connection_made()`` after its own "closed during connect" check; it + is part of the contract because socket-owning backends (asyncio, + Twisted) need it at connect time. + """ + sock = await _inet_create_connection(self, host, port, socket_options, + proxy_url=proxy_url, timeout_at=timeout_at) + if ssl is not None: + transport = KafkaSSLTransport(self, sock, ssl, host=host, + ssl_check_hostname=ssl_check_hostname) + else: + transport = KafkaTCPTransport(self, sock, host=host) + try: + await transport.handshake() + except Exception as e: + raise Errors.KafkaConnectionError('Handshake failed: %s' % e) + return transport + def sleep(self, delay): return KernelEvent('_sleep', delay) diff --git a/test/mock_broker.py b/test/mock_broker.py index cd61d355d..7c5b7f3d2 100644 --- a/test/mock_broker.py +++ b/test/mock_broker.py @@ -351,7 +351,7 @@ def stop(self, error=None): """Take the broker down, as if the process was killed. Aborts all live transports with ``error`` and refuses new connections - (an attached manager's ``_build_transport`` raises + (an attached manager's ``net.create_connection`` raises ``KafkaConnectionError``) until :meth:`start` is called. This exercises the client's real connection-lost, connect-failure, and reconnect-backoff paths. @@ -501,16 +501,16 @@ def attach(self, manager): """ broker = self - async def _mock_build_transport(node, timeout_at=None): + async def _mock_create_connection(protocol, host, port, **kwargs): if not broker.online: raise Errors.KafkaConnectionError( 'connect to %s:%s refused (MockBroker stopped)' - % (node.host, node.port)) + % (host, port)) return MockTransport( manager._net, broker, - node_id=node.node_id, host=node.host, port=node.port) + node_id=protocol.node_id, host=host, port=port) - manager._build_transport = _mock_build_transport + manager._net.create_connection = _mock_create_connection def client_factory(self): """Return a callable suitable for passing as ``kafka_client=...`` @@ -699,16 +699,16 @@ def attach(self, manager): """ cluster = self - async def _mock_build_transport(node, timeout_at=None): - broker = cluster._by_addr.get((node.host, node.port)) + async def _mock_create_connection(protocol, host, port, **kwargs): + broker = cluster._by_addr.get((host, port)) if broker is None or not broker.online: raise Errors.KafkaConnectionError( - 'connect to %s:%s refused' % (node.host, node.port)) + 'connect to %s:%s refused' % (host, port)) return MockTransport( manager._net, broker, - node_id=node.node_id, host=node.host, port=node.port) + node_id=protocol.node_id, host=host, port=port) - manager._build_transport = _mock_build_transport + manager._net.create_connection = _mock_create_connection def client_factory(self): """Return a callable suitable for passing as ``kafka_client=...`` to diff --git a/test/net/test_backend.py b/test/net/test_backend.py index f75df021a..2a2d50452 100644 --- a/test/net/test_backend.py +++ b/test/net/test_backend.py @@ -7,8 +7,9 @@ """ import threading -from kafka.net.backend import NetBackend +from kafka.net.backend import NetBackend, Transport from kafka.net.selector import NetworkSelector +from kafka.net.transport import KafkaTCPTransport # The full contract surface, kept here so a missing/renamed method fails loudly. @@ -16,10 +17,13 @@ 'start', 'stop', 'close', 'on_io_thread', 'call_soon', 'call_soon_threadsafe', 'call_soon_with_future', 'call_at', 'call_later', 'cancel', - 'sleep', 'wait_read', 'wait_write', - 'run', 'create_future', 'wakeup', 'unregister_event', + 'sleep', 'create_connection', + 'run', 'create_future', 'wakeup', ) +# Removed from the contract by the connection-seam revision (selector-private). +NON_CONTRACT_METHODS = ('wait_read', 'wait_write', 'unregister_event', 'poll') + class TestNetBackendContract: def test_networkselector_satisfies_protocol(self): @@ -40,10 +44,25 @@ def test_all_contract_methods_present_and_callable(self): for name in CONTRACT_METHODS: assert callable(getattr(net, name)), name - def test_poll_is_not_part_of_contract(self): - # poll() exists on NetworkSelector (legacy) but is intentionally not - # in the NetBackend surface; asyncio has no equivalent. - assert 'poll' not in CONTRACT_METHODS + def test_readiness_primitives_and_poll_excluded(self): + # wait_read/wait_write/unregister_event (selector-private, replaced by + # the connection seam) and legacy poll() are intentionally NOT in the + # contract, though they still exist on NetworkSelector. + net = NetworkSelector() + for name in NON_CONTRACT_METHODS: + assert name not in CONTRACT_METHODS + assert hasattr(net, name), name # still present on the selector impl + + +class TestTransportContract: + def test_kafkatcptransport_satisfies_transport(self): + # Method-presence check against the Transport protocol (no socket needed). + for name in ('write', 'close', 'abort', 'is_closing', + 'pause_reading', 'resume_reading', 'host_port'): + assert callable(getattr(KafkaTCPTransport, name)), name + + def test_plain_object_is_not_transport(self): + assert not isinstance(object(), Transport) class TestOnIoThread: diff --git a/test/net/test_manager.py b/test/net/test_manager.py index 036b6fa6a..fcc45f041 100644 --- a/test/net/test_manager.py +++ b/test/net/test_manager.py @@ -86,23 +86,23 @@ def test_proxy_url_takes_precedence_over_legacy(self, net): ) assert m.config['proxy_url'] == 'socks5://new:1080' - def test_build_transport_passes_proxy_url(self, net): - """_build_transport must forward the configured proxy_url to - create_connection. Regression guard against the kwarg name drifting - from the create_connection signature.""" - import asyncio + def test_connect_passes_proxy_url(self, net): + """_connect must forward the configured proxy_url to + net.create_connection. Regression guard against the kwarg name + drifting from the create_connection signature.""" m = KafkaConnectionManager(net, proxy_url='socks5://proxy:1080') - node = MagicMock(host='broker', port=9092) - async def fake_create_connection(*args, **kwargs): + node = MagicMock(host='broker', port=9092, node_id='bootstrap-0') + conn = KafkaConnection(net, node_id='bootstrap-0', **m.config) + + async def fake_create_connection(protocol, host, port, **kwargs): + # Close mid-connect so _connect short-circuits before + # connection_made()/initialize() -- we only assert the forwarded kwarg. + conn.close() return MagicMock() - with patch('kafka.net.manager.create_connection', - side_effect=fake_create_connection) as mc, \ - patch('kafka.net.manager.KafkaTCPTransport') as mock_transport_cls: - mock_transport = MagicMock() - mock_transport.handshake = MagicMock( - side_effect=lambda: asyncio.sleep(0)) - mock_transport_cls.return_value = mock_transport - net.run(m._build_transport(node)) + + with patch.object(net, 'create_connection', + side_effect=fake_create_connection) as mc: + net.run(m._connect(node, conn)) assert mc.call_args.kwargs.get('proxy_url') == 'socks5://proxy:1080' @@ -412,8 +412,8 @@ def test_close_no_connections(self, manager): class TestKafkaConnectionManagerConnectRace: """A connection can be closed (by manager.close() / bootstrap teardown) - while its _connect() coroutine is still awaiting _build_transport. When - the transport finally arrives, _connect must not resurrect the dead + while its _connect() coroutine is still awaiting net.create_connection. + When the transport finally arrives, _connect must not resurrect the dead connection via connection_made() -- doing so flips it back to `initializing`.""" @@ -423,13 +423,13 @@ def test_connect_discards_transport_when_closed_during_build(self, net): conn = KafkaConnection(net, node_id='bootstrap-0', **manager.config) transport = MagicMock() - async def fake_build_transport(n, timeout_at=None): + async def fake_create_connection(protocol, host, port, **kwargs): # Simulate a concurrent close landing mid-connect. conn.close() return transport - with patch.object(manager, '_build_transport', - side_effect=fake_build_transport): + with patch.object(net, 'create_connection', + side_effect=fake_create_connection): net.run(manager._connect(node, conn)) # Dead connection stays dead -- not resurrected to `initializing`. From 6afededd168b3651956d7aa91859b69de4943463 Mon Sep 17 00:00:00 2001 From: Dana Powers Date: Fri, 10 Jul 2026 16:26:30 -0700 Subject: [PATCH 3/5] Updates wrt SNI SSL fix on master; cache ssl_context in manager --- kafka/net/backend.py | 1 - kafka/net/manager.py | 5 ++-- kafka/net/selector.py | 7 ++--- kafka/net/transport.py | 17 ++++++------ test/net/test_transport.py | 56 ++++++++++++++++++++------------------ 5 files changed, 45 insertions(+), 41 deletions(-) diff --git a/kafka/net/backend.py b/kafka/net/backend.py index 630eed2f9..cabadb35e 100644 --- a/kafka/net/backend.py +++ b/kafka/net/backend.py @@ -173,7 +173,6 @@ async def create_connection( port: int, *, ssl: Any = None, - ssl_check_hostname: bool = True, proxy_url: Optional[str] = None, socket_options: Sequence[Any] = (), timeout_at: Optional[float] = None, diff --git a/kafka/net/manager.py b/kafka/net/manager.py index 220cef7cf..1e1d2d44a 100644 --- a/kafka/net/manager.py +++ b/kafka/net/manager.py @@ -9,6 +9,7 @@ from .metrics import KafkaManagerMetrics from kafka.cluster import ClusterMetadata import kafka.errors as Errors +from kafka.net.transport import KafkaSSLTransport from kafka.net.wakeup_notifier import WakeupNotifier from kafka.protocol.broker_version_data import BrokerVersionData from kafka.version import __version__ @@ -83,6 +84,7 @@ def __init__(self, net, **configs): self.cluster.attach(self) self._conns = {} self._backoff = dict() # node_id => (failures, backoff_until, socket_connect_setup_timeout_ms) + self.ssl_context = KafkaSSLTransport.build_ssl_context(self.config) if self.ssl_enabled else None # Cache the most recent SASL / SSL / auth failure per node so we can # surface it to the user instead of silently retrying forever. # Cleared on successful connect. @@ -209,8 +211,7 @@ async def _connect(self, node, conn, reset_backoff_on_connect=True, timeout_at=N try: transport = await self._net.create_connection( conn, node.host, node.port, - ssl=self._build_ssl_context() if self.ssl_enabled else None, - ssl_check_hostname=self.config['ssl_check_hostname'], + ssl=self.ssl_context, proxy_url=self.config['proxy_url'], socket_options=self.config['socket_options'], timeout_at=timeout_at) diff --git a/kafka/net/selector.py b/kafka/net/selector.py index 55cf685a6..7f11daa4f 100644 --- a/kafka/net/selector.py +++ b/kafka/net/selector.py @@ -488,8 +488,8 @@ def create_future(self): return SelectorFuture() async def create_connection(self, protocol, host, port, *, ssl=None, - ssl_check_hostname=True, proxy_url=None, - socket_options=(), timeout_at=None): + proxy_url=None, socket_options=(), + timeout_at=None): """Establish and return a connected transport to host:port. The selector owns the raw socket: DNS + non-blocking connect (with @@ -503,8 +503,7 @@ async def create_connection(self, protocol, host, port, *, ssl=None, sock = await _inet_create_connection(self, host, port, socket_options, proxy_url=proxy_url, timeout_at=timeout_at) if ssl is not None: - transport = KafkaSSLTransport(self, sock, ssl, host=host, - ssl_check_hostname=ssl_check_hostname) + transport = KafkaSSLTransport(self, sock, ssl, host=host) else: transport = KafkaTCPTransport(self, sock, host=host) try: diff --git a/kafka/net/transport.py b/kafka/net/transport.py index 5a5365753..f44d0ff20 100644 --- a/kafka/net/transport.py +++ b/kafka/net/transport.py @@ -381,20 +381,21 @@ class KafkaSSLTransport(KafkaTCPTransport): 'ssl_password': None, 'ssl_crlfile': None, } - def __init__(self, net, sock, host=None, **configs): - self.ssl_config = copy.copy(self.DEFAULT_CONFIG) - for key in self.ssl_config: - if key in configs: - self.ssl_config[key] = configs[key] - self._ssl_context = self._build_ssl_context(self.ssl_config) + def __init__(self, net, sock, ssl_context, host=None): + self._ssl_context = ssl_context server_hostname = host.rstrip('.') if host is not None else None sock = self._ssl_context.wrap_socket( sock, server_hostname=server_hostname, do_handshake_on_connect=False) super().__init__(net, sock, host=host) - @staticmethod - def _build_ssl_context(config): + @classmethod + def build_ssl_context(cls, configs): + config = copy.copy(cls.DEFAULT_CONFIG) + for key in config: + if key in configs: + config[key] = configs[key] + if config['ssl_context'] is not None: return config['ssl_context'] ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT) diff --git a/test/net/test_transport.py b/test/net/test_transport.py index d1ed0ba90..32e4651a6 100644 --- a/test/net/test_transport.py +++ b/test/net/test_transport.py @@ -1,4 +1,5 @@ import socket +import ssl import time from unittest.mock import MagicMock @@ -385,72 +386,75 @@ def _make_ssl_sock(self): def test_sni_sent_when_check_hostname_true(self, net): sock, ctx, _ = self._make_ssl_sock() - KafkaSSLTransport(net, sock, host='broker.example.com', - ssl_context=ctx, ssl_check_hostname=True) + ctx.check_hostname = True + KafkaSSLTransport(net, sock, ctx, host='broker.example.com') _, kwargs = ctx.wrap_socket.call_args assert kwargs['server_hostname'] == 'broker.example.com' def test_sni_sent_when_check_hostname_false(self, net): # The bug: SNI used to be suppressed when verification was disabled. sock, ctx, _ = self._make_ssl_sock() - KafkaSSLTransport(net, sock, host='broker.example.com', - ssl_context=ctx, ssl_check_hostname=False) + ctx.check_hostname = False + KafkaSSLTransport(net, sock, ctx, host='broker.example.com') _, kwargs = ctx.wrap_socket.call_args assert kwargs['server_hostname'] == 'broker.example.com' def test_sni_strips_trailing_dot(self, net): # A trailing dot is a valid FQDN but illegal in the SNI extension. sock, ctx, _ = self._make_ssl_sock() - KafkaSSLTransport(net, sock, host='broker.example.com.', - ssl_context=ctx, ssl_check_hostname=False) + ctx.check_hostname = False + KafkaSSLTransport(net, sock, ctx, host='broker.example.com.') _, kwargs = ctx.wrap_socket.call_args assert kwargs['server_hostname'] == 'broker.example.com' def test_sni_none_when_host_missing(self, net): sock, ctx, _ = self._make_ssl_sock() - KafkaSSLTransport(net, sock, host=None, ssl_context=ctx) + KafkaSSLTransport(net, sock, ctx, host=None) _, kwargs = ctx.wrap_socket.call_args assert kwargs['server_hostname'] is None def test_handshake_not_done_on_connect(self, net): sock, ctx, _ = self._make_ssl_sock() - KafkaSSLTransport(net, sock, host='broker.example.com', ssl_context=ctx) + KafkaSSLTransport(net, sock, ctx, host='broker.example.com') _, kwargs = ctx.wrap_socket.call_args assert kwargs['do_handshake_on_connect'] is False def test_provided_ssl_context_is_used(self, net): sock, ctx, wrapped = self._make_ssl_sock() - t = KafkaSSLTransport(net, sock, host='broker.example.com', - ssl_context=ctx) + t = KafkaSSLTransport(net, sock, ctx, host='broker.example.com') assert t._ssl_context is ctx assert t._sock is wrapped - def test_config_defaults_and_overrides(self, net): - sock, ctx, _ = self._make_ssl_sock() - t = KafkaSSLTransport(net, sock, host='h', ssl_context=ctx, - ssl_check_hostname=False) - # Explicitly-passed ssl_* keys land in ssl_config... - assert t.ssl_config['ssl_check_hostname'] is False - assert t.ssl_config['ssl_context'] is ctx - # ...unspecified keys keep their defaults. - assert t.ssl_config['ssl_cafile'] is None - assert t.ssl_config['ssl_check_hostname'] is not None + def test_ssl_context_is_required(self, net): + # The transport no longer builds a context itself; callers must pass + # a pre-built one (via build_ssl_context). Omitting it is a TypeError, + # not a silently-default context. + sock, _, _ = self._make_ssl_sock() + with pytest.raises(TypeError): + KafkaSSLTransport(net, sock, host='broker.example.com') class TestBuildSSLContext: def test_returns_provided_context(self): ctx = MagicMock() - config = dict(KafkaSSLTransport.DEFAULT_CONFIG, ssl_context=ctx) - assert KafkaSSLTransport._build_ssl_context(config) is ctx + config = dict(ssl_context=ctx) + assert KafkaSSLTransport.build_ssl_context(config) is ctx def test_check_hostname_propagates_to_context(self): - config = dict(KafkaSSLTransport.DEFAULT_CONFIG, ssl_check_hostname=False) - ctx = KafkaSSLTransport._build_ssl_context(config) + config = dict(ssl_check_hostname=False) + ctx = KafkaSSLTransport.build_ssl_context(config) assert ctx.check_hostname is False def test_check_hostname_true_requires_verification(self): - config = dict(KafkaSSLTransport.DEFAULT_CONFIG, ssl_check_hostname=True) - ctx = KafkaSSLTransport._build_ssl_context(config) + config = dict(ssl_check_hostname=True) + ctx = KafkaSSLTransport.build_ssl_context(config) + assert ctx.check_hostname is True + + def test_empty_config_builds_default_context(self): + # Missing keys fall back to DEFAULT_CONFIG: a real TLS-client context + # with hostname checking on. This is the default-build path. + ctx = KafkaSSLTransport.build_ssl_context({}) + assert isinstance(ctx, ssl.SSLContext) assert ctx.check_hostname is True From 6c0ad5a1f3ba83dda9fa866297095a5669685fb3 Mon Sep 17 00:00:00 2001 From: Dana Powers Date: Fri, 10 Jul 2026 16:51:54 -0700 Subject: [PATCH 4/5] merge conflict --- kafka/net/backend.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/kafka/net/backend.py b/kafka/net/backend.py index cabadb35e..7f4c57b47 100644 --- a/kafka/net/backend.py +++ b/kafka/net/backend.py @@ -46,7 +46,7 @@ * **Future factory** -- ``create_future`` (see ``BackendFuture``). * **Cross-thread wake** -- ``wakeup``. """ -from typing import Any, Optional, Protocol, Sequence, Tuple, runtime_checkable +from typing import Any, Callable, Optional, Protocol, Sequence, Tuple, runtime_checkable @runtime_checkable From 3232f6f2403dd5b9f7301d7c52ecafd4d975e315 Mon Sep 17 00:00:00 2001 From: Dana Powers Date: Fri, 10 Jul 2026 16:52:01 -0700 Subject: [PATCH 5/5] ignore pylint --- test/net/test_transport.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/test/net/test_transport.py b/test/net/test_transport.py index 32e4651a6..36af1abc0 100644 --- a/test/net/test_transport.py +++ b/test/net/test_transport.py @@ -431,7 +431,7 @@ def test_ssl_context_is_required(self, net): # not a silently-default context. sock, _, _ = self._make_ssl_sock() with pytest.raises(TypeError): - KafkaSSLTransport(net, sock, host='broker.example.com') + KafkaSSLTransport(net, sock, host='broker.example.com') # pylint: disable=E1120 class TestBuildSSLContext: