net: define the NetBackend loop-surface contract

kafka/net/backend.py

@@ -1,11 +1,52 @@
-"""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.
+
+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.
+* ``register_event`` / selector internals -- backend-private plumbing.
+
+Method families:
+
+* **Lifecycle** -- ``start`` / ``stop`` / ``close`` / ``on_io_thread``.
+* **Scheduling** -- ``call_soon`` / ``call_soon_threadsafe`` /
+  ``call_soon_with_future`` / ``call_at`` / ``call_later`` / ``cancel``.
+* **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, Protocol, runtime_checkable
+from typing import Any, Callable, Optional, Protocol, Sequence, Tuple, runtime_checkable
 
 
 @runtime_checkable
@@ -59,3 +100,106 @@ 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 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.
+
+    ``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_*."""
+
+    # --- timing (core coroutines await this) ------------------------------
+    def sleep(self, delay: float) -> Any:
+        """Awaitable that resolves after ``delay`` seconds."""
+
+    # --- connection seam --------------------------------------------------
+    async def create_connection(
+        self,
+        protocol: Any,
+        host: str,
+        port: int,
+        *,
+        ssl: Any = None,
+        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:
+        """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."""

kafka/net/manager.py

@@ -5,12 +5,11 @@
 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.transport import KafkaSSLTransport
 from kafka.net.wakeup_notifier import WakeupNotifier
 from kafka.protocol.broker_version_data import BrokerVersionData
 from kafka.version import __version__
@@ -85,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.
@@ -203,33 +203,18 @@ 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.ssl_context,
+                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

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__
 
 
@@ -277,6 +279,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.
 
@@ -475,6 +487,31 @@ def create_future(self):
         """
         return SelectorFuture()
 
+    async def create_connection(self, protocol, host, port, *, ssl=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
+        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)
+        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)
 

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)

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"

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