Add backend e2e tests against both selector + asyncio

.github/workflows/python-package.yml

@@ -18,7 +18,7 @@ jobs:
   build:
 
     runs-on: ubuntu-latest
-    name: "Test: python ${{ matrix.python }} / kafka ${{ matrix.kafka }}"
+    name: "py ${{ matrix.python }} / kafka ${{ matrix.kafka }} / net ${{ matrix.net }}"
     continue-on-error: ${{ matrix.experimental || false }}
     strategy:
       fail-fast: false
@@ -34,25 +34,36 @@ jobs:
           - "3.0.2"
           - "3.5.2"
           - "3.9.0"
-          - "4.3.0"
         python:
           - "3.14"
+        net:
+          - "random"
         include:
-          #- python: "pypy3.9"
-          #  kafka: "2.6.0"
-          #  experimental: true
           - python: "3.8"
             kafka: "4.3.0"
+            net: "random"
           - python: "3.9"
             kafka: "4.3.0"
+            net: "random"
           - python: "3.10"
             kafka: "4.3.0"
+            net: "random"
           - python: "3.11"
             kafka: "4.3.0"
+            net: "random"
           - python: "3.12"
             kafka: "4.3.0"
+            net: "random"
           - python: "3.13"
             kafka: "4.3.0"
+            net: "random"
+          - python: "3.14"
+            kafka: "4.3.0"
+            net: "asyncio"
+          - python: "3.14"
+            kafka: "4.3.0"
+            net: "selector"
+
 
     steps:
       - uses: actions/checkout@v7
@@ -92,17 +103,20 @@ jobs:
           key: ${{ steps.cache-servers-dist-restore.outputs.cache-primary-key }}
       - name: Pytest
         # coverage runs all tests, so we can skip this test step if we're going to do coverage
-        if: matrix.python != '3.14' || matrix.kafka != '4.0.0'
+        # (skip only the exact coverage job -- negation of the coverage `if` below)
+        if: matrix.python != '3.14' || matrix.kafka != '4.3.0' || matrix.net != 'selector'
         run: make test
         env:
           KAFKA_VERSION: ${{ matrix.kafka }}
+          KAFKA_PYTHON_NET: ${{ matrix.net || '' }}
       - name: Pytest (with coverage)
-        if: matrix.python == '3.14' && matrix.kafka == '4.0.0'
+        if: matrix.python == '3.14' && matrix.kafka == '4.3.0' && matrix.net == 'selector'
         run: make test-coverage
         env:
           KAFKA_VERSION: ${{ matrix.kafka }}
+          KAFKA_PYTHON_NET: ${{ matrix.net || '' }}
       - name: Upload coverage report
-        if: matrix.python == '3.14' && matrix.kafka == '4.0.0'
+        if: matrix.python == '3.14' && matrix.kafka == '4.3.0' && matrix.net == 'selector'
         uses: actions/upload-artifact@v7
         with:
           name: coverage-report

kafka/net/backend/__init__.py

@@ -1,6 +1,6 @@
 from .abstract import (
     NetBackend, NetTransport, NetProtocol, NetBackendFuture,
-    resolve_backend, register_backend_lazy,
+    list_backends, resolve_backend, register_backend_lazy,
 )
 
 register_backend_lazy('selector', 'kafka.net.backend.selector', 'NetworkSelector')

kafka/net/backend/abstract.py

@@ -258,6 +258,11 @@ def wakeup(self) -> None:
 _BACKENDS = {}
 
 
+def list_backends():
+    """List all registered backends."""
+    return list(_BACKENDS.keys())
+
+
 def register_backend(name, factory):
     """Register a named backend factory for ``net='<name>'`` selection."""
     _BACKENDS[name] = factory

kafka/net/backend/asyncio_backend.py

@@ -10,12 +10,17 @@
 thread on the loop thread; it does not run on the caller's own loop.
 """
 import asyncio
+import copy
 import inspect
+import logging
 import threading
 import time
 
 import kafka.errors as Errors
 from kafka.future import Future
+from kafka.version import __version__
+
+log = logging.getLogger(__name__)
 
 
 class AsyncioFuture(Future):
@@ -78,7 +83,28 @@ def cancel(self):
 
 
 class AsyncioBackend:
+    DEFAULT_CONFIG = {
+        'client_id': 'kafka-python-' + __version__,
+        # Default operation deadline for a cross-thread run() call that does not
+        # pass its own timeout_ms. Bounds the caller's blocking wait so a stalled
+        # IO loop (e.g. a synchronous rebalance listener/assignor or a blocking
+        # DNS lookup on the IO thread) cannot hang the caller indefinitely.
+        # Mirrors the Java client's default.api.timeout.ms.
+        'default_api_timeout_ms': 60000,
+        # Extra slack added on top of the operation deadline before run()'s
+        # cross-thread wait gives up. The coroutine enforces the operation
+        # deadline itself on a healthy loop; this margin ensures that self-timeout
+        # (and its unwind/retry-backoff) resolves and wins the race, so the
+        # backstop only trips when the loop genuinely isn't running the coroutine.
+        'bridge_grace_ms': 5000,
+    }
+
     def __init__(self, *, loop=None, loop_factory=None, **configs):
+        self.config = copy.copy(self.DEFAULT_CONFIG)
+        for key in self.config:
+            if key in configs:
+                self.config[key] = configs[key]
+
         # Loop acquisition is a strategy, not a hardcode (forward seam for a
         # native/Phase-2 mode). Default: own a fresh loop on our daemon thread.
         # loop_factory lets callers pick the loop implementation (e.g. uvloop)
@@ -93,13 +119,6 @@ def __init__(self, *, loop=None, loop_factory=None, **configs):
             self._owns_loop = True
         self._io_thread = None
         self._closed = False
-        self._client_id = configs.get('client_id') or 'kafka-python'
-        # See NetworkSelector: default operation deadline + grace margin for the
-        # cross-thread run() liveness backstop (#3121).
-        self._default_api_timeout_ms = configs.get('default_api_timeout_ms') or 60000
-        self._bridge_grace_ms = configs.get('bridge_grace_ms')
-        if self._bridge_grace_ms is None:
-            self._bridge_grace_ms = 5000
         # Strong refs to live tasks (asyncio only holds weak refs, so bare
         # tasks can be GC'd mid-flight); mirrors NetworkSelector._pending_tasks.
         self._pending = set()
@@ -112,7 +131,7 @@ def start(self):
         if self._io_thread is not None:
             return
         t = threading.Thread(target=self._run_forever,
-                             name='kafka-io-%s' % self._client_id, daemon=True)
+                             name='kafka-io-%s' % self.config['client_id'], daemon=True)
         self._io_thread = t
         t.start()
 
@@ -159,6 +178,33 @@ def _fail_pending_waiters(self, exc):
                 state['exception'] = exc
             event.set()
 
+    def _bridge_deadline_secs(self, timeout_ms):
+        """Wall-clock ceiling for a cross-thread run() wait, in seconds.
+
+        The operation deadline defaults to ``default_api_timeout_ms`` when the
+        caller passes ``timeout_ms=None``; ``bridge_grace_ms`` is added so the
+        coroutine's own (equal) deadline wins the race on a healthy loop.
+        """
+        op_ms = timeout_ms if timeout_ms is not None else self.config['default_api_timeout_ms']
+        if op_ms >= threading.TIMEOUT_MAX:
+            return None
+        return (op_ms + self.config['bridge_grace_ms']) / 1000
+
+    def _bridge_timeout(self, coro, timeout_ms):
+        op_ms = timeout_ms if timeout_ms is not None else self.config['default_api_timeout_ms']
+        name = getattr(coro, '__name__', None) or repr(coro)
+        exc = Errors.KafkaTimeoutError(
+            'net.run(%s) did not complete within %d ms (+%d ms grace). The IO '
+            'event loop may be stalled by blocking work on the IO thread '
+            '(e.g. a synchronous rebalance listener/assignor).'
+            % (name, op_ms, self.config['bridge_grace_ms']))
+        # A caught KafkaTimeoutError is otherwise invisible; surface the stall so
+        # operators can see it. The coroutine keeps running until it completes or
+        # hits its own deadline, so any side effects (e.g. an offset commit) may
+        # still take effect (see the late-completion warning in run()).
+        log.warning('%s', exc)
+        return exc
+
     # --- scheduling -------------------------------------------------------
     def _spawn(self, coro):
         task = self._loop.create_task(coro)
@@ -305,8 +351,7 @@ def run(self, coro, *args, timeout_ms=None):
                 "(or another IO-thread callback) calls a blocking consumer/admin API. "
                 "Use AsyncConsumerRebalanceListener and await the async variant, "
                 "or move the blocking work to a worker thread.")
-        op_ms = timeout_ms if timeout_ms is not None else self._default_api_timeout_ms
-        deadline_secs = (op_ms + self._bridge_grace_ms) / 1000
+        deadline_secs = self._bridge_deadline_secs(timeout_ms)
         event = threading.Event()
         state = {'value': None, 'exception': None}
 
@@ -325,14 +370,7 @@ async def waiter():
             self._pending_waiters[event] = state
         self.call_soon(waiter)
         if not event.wait(timeout=deadline_secs):
-            # Loop never ran the coroutine to completion; leave the waiter
-            # registered (its finally pops it) and surface a liveness timeout.
-            name = getattr(coro, '__name__', None) or repr(coro)
-            raise Errors.KafkaTimeoutError(
-                'net.run(%s) did not complete within %d ms (+%d ms grace). The '
-                'IO event loop may be stalled by blocking work on the IO thread '
-                '(e.g. a synchronous rebalance listener/assignor).'
-                % (name, op_ms, self._bridge_grace_ms))
+            raise self._bridge_timeout(coro, timeout_ms)
         if state['exception'] is not None:
             raise state['exception']  # pylint: disable=raising-bad-type
         return state['value']

test/conftest.py

@@ -65,6 +65,30 @@ def net():
         backend.close()
 
 
+@pytest.fixture(params=['selector', 'asyncio'])
+def all_net(request):
+    """A *started* backend, parametrized over selector + asyncio, for curated
+    both-backends end-to-end tests.
+
+    Unlike the default ``net`` fixture (an unstarted NetworkSelector that tests
+    drive inline), this runs the loop on its own IO thread -- required for the
+    asyncio backend, whose ``run()`` has no inline no-IO-thread fallback. Pair
+    it with a MockCluster (stateful): a started loop runs background heartbeat /
+    metadata-refresh coroutines that fire unscripted requests, which MockCluster
+    answers and a scripted MockBroker would not.
+
+    The manager does not own a passed-in instance, so it neither starts nor
+    closes it -- this fixture owns the lifecycle and closes it on teardown.
+    """
+    from kafka.net.backend import resolve_backend
+    backend = resolve_backend(request.param, {})
+    backend.start()
+    try:
+        yield backend
+    finally:
+        backend.close()
+
+
 @pytest.fixture
 def manager(net, broker):
     broker.attach(net)

test/integration/fixtures.py

@@ -1,10 +1,13 @@
 import atexit
 import base64
+import hashlib
 import logging
+import random
 import os
 import os.path
 import socket
 import subprocess
+import sys
 import time
 from urllib.parse import urlparse
 import uuid
@@ -13,12 +16,77 @@
 
 from kafka import errors, KafkaAdminClient
 from kafka.errors import InvalidReplicationFactorError
+from kafka.net.backend import list_backends
 from test.testutil import env_kafka_version, random_string
 from test.service import ExternalService, SpawnedService
 
 log = logging.getLogger(__name__)
 
 
+_UNRESOLVED = object()
+_selected_net_cache = _UNRESOLVED
+
+
+def _git_revision():
+    """Commit SHA for deterministic backend selection: ``GITHUB_SHA`` in CI,
+    else ``git rev-parse HEAD``, else None (e.g. a source tarball with no git)."""
+    sha = os.environ.get('GITHUB_SHA')
+    if sha:
+        return sha
+    try:
+        return subprocess.check_output(
+            ['git', 'rev-parse', 'HEAD'],
+            cwd=os.path.dirname(os.path.abspath(__file__)),
+            stderr=subprocess.DEVNULL).decode().strip()
+    except (OSError, subprocess.SubprocessError):
+        return None
+
+
+def _spread_net(sha=None, python=None, kafka=None):
+    """Deterministically pick a backend, spread across the CI matrix.
+
+    Seeds the choice with the commit SHA *and* the per-job matrix variables
+    (Python major.minor + KAFKA_VERSION). Every job in one CI run shares the
+    same ``GITHUB_SHA``, so the matrix vars are what fan the ~17 jobs out across
+    backends within that run; the SHA reshuffles the assignment on later commits,
+    so each (python, kafka) cell covers every backend over time. A given
+    (commit, python, kafka) always picks the same backend -- reproducible on
+    re-run. ``hashlib`` (not the salted builtin ``hash()``) keeps it stable
+    across processes.
+    """
+    backends = sorted(list_backends())
+    sha = (_git_revision() or 'no-sha') if sha is None else sha
+    python = ('%d.%d' % sys.version_info[:2]) if python is None else python
+    kafka = os.environ.get('KAFKA_VERSION', '') if kafka is None else kafka
+    seed = '%s:%s:%s' % (sha, python, kafka)
+    idx = int(hashlib.sha256(seed.encode()).hexdigest(), 16) % len(backends)
+    net = backends[idx]
+    log.warning('KAFKA_PYTHON_NET=random -> net=%r (seed: sha=%s python=%s '
+                'kafka=%s; re-run with KAFKA_PYTHON_NET=%s to reproduce)',
+                net, sha[:12], python, kafka or '-', net)
+    return net
+
+
+def _selected_net():
+    """Resolve the KAFKA_PYTHON_NET override once per run, and log the choice.
+
+    ``KAFKA_PYTHON_NET=random`` picks a backend *once* (not per client) so the
+    whole run uses one backend, spread across the matrix (see ``_spread_net``).
+    Any other value pins that backend. Returns the backend name, or '' when
+    unset (default selector).
+    """
+    global _selected_net_cache
+    if _selected_net_cache is _UNRESOLVED:
+        net = os.environ.get('KAFKA_PYTHON_NET') or ''
+        if net == 'random':
+            net = _spread_net()
+        elif net:
+            log.info('KAFKA_PYTHON_NET=%r: running against the %r net backend',
+                     net, net)
+        _selected_net_cache = net
+    return _selected_net_cache
+
+
 def create_topics(broker, topic_names, num_partitions=None, replication_factor=None):
     """Create topics on the given broker fixture.
 
@@ -729,13 +797,12 @@ def _enrich_client_params(self, params, **defaults):
         # Run the whole integration suite against a chosen net backend, e.g.
         # KAFKA_PYTHON_NET=asyncio make test. Unset -> default (selector).
         # setdefault so a test that pins net= still wins.
-        net = os.environ.get('KAFKA_PYTHON_NET')
+        net = _selected_net()
         if net:
             params.setdefault('net', net)
         return params
 
 
-
 def get_api_versions():
     logging.basicConfig(level=logging.ERROR)
     k = KafkaFixture.instance(0)