diff --git a/.github/workflows/python-package.yml b/.github/workflows/python-package.yml index 3def1a2e5..639fb390f 100644 --- a/.github/workflows/python-package.yml +++ b/.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 diff --git a/kafka/net/backend/__init__.py b/kafka/net/backend/__init__.py index fedff3730..9546a3cde 100644 --- a/kafka/net/backend/__init__.py +++ b/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') diff --git a/kafka/net/backend/abstract.py b/kafka/net/backend/abstract.py index e6d657b69..633775c50 100644 --- a/kafka/net/backend/abstract.py +++ b/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=''`` selection.""" _BACKENDS[name] = factory diff --git a/kafka/net/backend/asyncio_backend.py b/kafka/net/backend/asyncio_backend.py index 7eeb074ac..6bfd62dcc 100644 --- a/kafka/net/backend/asyncio_backend.py +++ b/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'] diff --git a/test/conftest.py b/test/conftest.py index 111af8d5b..c24b42550 100644 --- a/test/conftest.py +++ b/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) diff --git a/test/integration/fixtures.py b/test/integration/fixtures.py index 8ca5a5cf3..c6f018de5 100644 --- a/test/integration/fixtures.py +++ b/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) diff --git a/test/net/backend/test_all_backends.py b/test/net/backend/test_all_backends.py new file mode 100644 index 000000000..fd8d0df5b --- /dev/null +++ b/test/net/backend/test_all_backends.py @@ -0,0 +1,104 @@ +"""Curated both-backends end-to-end coverage. + +A small, high-signal set of full-stack round-trips (bootstrap, send, metadata +refresh, consumer-group join) run against BOTH the selector and asyncio backends +via the parametrized ``all_net`` fixture. This is the cross-backend counterpart +to the per-backend suites in this package: those exercise each backend in +isolation; these prove the manager / coordinator async bridge (run / wait_for / +send / bootstrap) behaves identically on both. + +Driven through MockCluster (stateful), not a scripted MockBroker: a started loop +runs background heartbeat / metadata-refresh coroutines that fire unscripted +requests, which MockCluster answers and a scripted MockBroker would not. +""" +from kafka.consumer.subscription_state import SubscriptionState +from kafka.coordinator.consumer import ConsumerCoordinator +from kafka.net.compat import KafkaNetClient +from kafka.net.manager import KafkaConnectionManager +from kafka.protocol.metadata import MetadataRequest, MetadataResponse +from kafka.structs import TopicPartition +from test.mock_broker import MockCluster + + +def _metadata_topic(name, num_partitions, node_id=0): + """A MetadataResponseTopic for MockCluster.set_metadata(topics=[...]).""" + Topic = MetadataResponse.MetadataResponseTopic + Partition = Topic.MetadataResponsePartition + return Topic(version=8, error_code=0, name=name, is_internal=False, + partitions=[ + Partition(version=8, error_code=0, partition_index=p, + leader_id=node_id, leader_epoch=0, + replica_nodes=[node_id], isr_nodes=[node_id], + offline_replicas=[]) + for p in range(num_partitions)]) + + +def _manager(all_net, mock_cluster): + manager = KafkaConnectionManager( + all_net, bootstrap_servers=mock_cluster.bootstrap_servers(), + api_version=mock_cluster.broker_version, request_timeout_ms=5000) + mock_cluster.attach(manager) + return manager + + +def test_bootstrap_and_metadata_send(all_net): + """Bootstrap + a MetadataRequest round-trip through the manager.""" + cluster = MockCluster(num_brokers=1) + cluster.set_metadata(topics=[_metadata_topic('t', num_partitions=2)]) + manager = _manager(all_net, cluster) + try: + manager.bootstrap(timeout_ms=5000) + assert manager.bootstrapped + assert manager.cluster.brokers() + + async def do_send(): + return await manager.send(MetadataRequest[0]([])) + resp = all_net.run(do_send) + assert resp is not None + finally: + manager.close() + + +def test_metadata_refresh(all_net): + """cluster.refresh_metadata() drives a real MetadataRequest and applies it.""" + cluster = MockCluster(num_brokers=1) + cluster.set_metadata(topics=[_metadata_topic('t', num_partitions=3)]) + manager = _manager(all_net, cluster) + try: + manager.bootstrap(timeout_ms=5000) + all_net.run(manager.cluster.refresh_metadata) + assert 't' in manager.cluster.topics() + assert manager.cluster.partitions_for_topic('t') == {0, 1, 2} + finally: + manager.close() + + +def test_consumer_group_join_assigns_partitions(all_net, metrics): + """A consumer joins a MockCluster group, is elected leader, runs the + assignor, and is assigned every partition -- exercising find-coordinator + + join + sync + heartbeat on both backends.""" + cluster = MockCluster(num_brokers=1) + cluster.set_metadata(topics=[_metadata_topic('t', num_partitions=3)]) + group = cluster.add_group('grp', coordinator=0) + manager = _manager(all_net, cluster) + client = KafkaNetClient(net=all_net, manager=manager) + coordinator = ConsumerCoordinator( + client, SubscriptionState(), metrics=metrics, + api_version=cluster.broker_version, + heartbeat_interval_ms=20, retry_backoff_ms=20, group_id='grp') + manager.bootstrap(timeout_ms=5000) + try: + coordinator._subscription.subscribe(topics=['t']) + client.cluster.request_update() + + assert coordinator.ensure_active_group(timeout_ms=5000) + assert coordinator._subscription.assigned_partitions() == { + TopicPartition('t', 0), TopicPartition('t', 1), TopicPartition('t', 2)} + assert coordinator._is_leader + assert group.generation == 1 + finally: + coordinator._close_heartbeat() + coordinator.reset_generation() + coordinator.close(timeout_ms=0) + client.close() + manager.close()