From d790a19fe4131f1d36d46a9d434f0626817c75d6 Mon Sep 17 00:00:00 2001 From: Dana Powers Date: Mon, 29 Jun 2026 21:34:41 +0000 Subject: [PATCH 1/8] net: add create_future() seam for pluggable backends Introduce NetworkSelector.create_future() (and a KafkaConnectionManager forwarder) as the portability seam for an upcoming pluggable async backend: core coroutines that await a Future while it is pending must obtain it from the backend so an alternate backend (asyncio, Twisted) can return its own awaitable type instead of kafka.future.Future. Route the loop-created, awaited-while-pending Future constructions through create_future(): - KafkaConnection._init_future (awaited via __await__), send_request and _send_request response futures, SaslReauthenticator._drain_future - ClusterMetadata._refresh_future - KafkaConnectionManager.wait_for wrapper - WakeupNotifier._fut Left as plain Future: pre-resolved futures (success()/failure() never yield), callback-only futures (_close_future), and user-thread-created futures bridged into the loop via wait_for's wrapper (cluster update future, fetcher wakeup). Behavior is identical under NetworkSelector (create_future() returns Future()); full unit suite green. Co-Authored-By: Claude Opus 4.8 (1M context) --- kafka/cluster.py | 2 +- kafka/net/connection.py | 10 +++---- kafka/net/manager.py | 10 ++++++- kafka/net/selector.py | 10 +++++++ kafka/net/wakeup_notifier.py | 4 +-- test/net/test_manager.py | 6 ++++ test/net/test_selector.py | 54 ++++++++++++++++++++++++++++++++++++ 7 files changed, 86 insertions(+), 10 deletions(-) diff --git a/kafka/cluster.py b/kafka/cluster.py index e41ac59eb..93fdf9bcb 100644 --- a/kafka/cluster.py +++ b/kafka/cluster.py @@ -150,7 +150,7 @@ async def refresh_metadata(self, node_id=None): log.debug('Metadata refresh already in flight; awaiting existing') await self._refresh_future return - self._refresh_future = Future() + self._refresh_future = self._manager.create_future() try: await self._do_refresh_metadata(node_id) except Exception as exc: diff --git a/kafka/net/connection.py b/kafka/net/connection.py index 2794e1a22..5b1a67377 100644 --- a/kafka/net/connection.py +++ b/kafka/net/connection.py @@ -53,8 +53,8 @@ def __init__(self, net, node_id=None, broker_version_data=None, **configs): self.paused = set() self.connected = False self.initializing = True - self._init_future = Future() - self._close_future = Future() + self._init_future = net.create_future() # awaited via __await__; backend-native + self._close_future = Future() # callback-only, never awaited self.in_flight_requests = collections.deque() self.broker_version_data = broker_version_data self._api_versions_idx = ApiVersionsRequest.max_version # version of ApiVersionsRequest to try on first connect @@ -116,7 +116,7 @@ def _timeout_at(self, now=None, timeout_ms=None): return now + self._timeout_secs def send_request(self, request, request_timeout_ms=None): - future = Future() + future = self.net.create_future() timeout_at = self._timeout_at(timeout_ms=request_timeout_ms) if self.initializing or self._reauth.is_reauthenticating: self._request_buffer.append((request, future, timeout_at)) @@ -131,7 +131,7 @@ def send_request(self, request, request_timeout_ms=None): def _send_request(self, request, future=None, timeout_at=None): if future is None: - future = Future() + future = self.net.create_future() if self.closed: return future.failure(Errors.KafkaConnectionError('closed')) if request.API_VERSION is None: @@ -587,7 +587,7 @@ async def _do_reauth(self): # reasoning about FIFO interleaving with the broker's reauth # validation). while self._conn.in_flight_requests and not self._conn.closed: - self._drain_future = Future() + self._drain_future = self._conn.net.create_future() if not self._conn.in_flight_requests: break await self._drain_future diff --git a/kafka/net/manager.py b/kafka/net/manager.py index 850a5938a..ab034a4d3 100644 --- a/kafka/net/manager.py +++ b/kafka/net/manager.py @@ -412,7 +412,7 @@ async def wait_for(self, future, timeout_ms): """ if timeout_ms is None: return await future - wrapper = Future() + wrapper = self._net.create_future() def _on_success(value): if not wrapper.is_done: wrapper.success(value) @@ -431,6 +431,14 @@ def _on_timeout(): finally: self._net.cancel(timer) + def create_future(self): + """Create a Future suitable for awaiting on the underlying loop. + + Forwards to the backend so loop coroutines get the backend's native + awaitable type. See ``NetworkSelector.create_future``. + """ + return self._net.create_future() + def call_soon(self, coro, *args): """Accepts a coroutine / awaitable / function and schedules it on the event loop. diff --git a/kafka/net/selector.py b/kafka/net/selector.py index 5c2a731bb..384d93322 100644 --- a/kafka/net/selector.py +++ b/kafka/net/selector.py @@ -445,6 +445,16 @@ def reschedule(self, when, task): self.call_at(when, task) return task + def create_future(self): + """Create a Future suitable for awaiting on this backend's loop. + + Portability seam for pluggable backends: core coroutines call this + instead of constructing ``Future`` directly, so an alternate backend + (asyncio, Twisted) can return its own awaitable type. The selector's + native awaitable is ``kafka.future.Future``. + """ + return Future() + def sleep(self, delay): return KernelEvent('_sleep', delay) diff --git a/kafka/net/wakeup_notifier.py b/kafka/net/wakeup_notifier.py index 67041deb5..fed4d655e 100644 --- a/kafka/net/wakeup_notifier.py +++ b/kafka/net/wakeup_notifier.py @@ -1,7 +1,5 @@ import weakref -from kafka.future import Future - class WakeupNotifier: """await wakeup(timeout_secs) when either ``timeout_secs`` elapses or @@ -49,7 +47,7 @@ def _wakeup(self): async def __call__(self, timeout_secs=None): if self._fut is not None: raise RuntimeError('Concurrent access to WakeupNotifier!') - self._fut = Future() + self._fut = self._net.create_future() if self._pending: self._pending = False self._fut.success(None) diff --git a/test/net/test_manager.py b/test/net/test_manager.py index 33d59f642..5d767c367 100644 --- a/test/net/test_manager.py +++ b/test/net/test_manager.py @@ -43,6 +43,12 @@ def test_api_versions(self, net): m = KafkaConnectionManager(net, api_version=(1, 0)) assert m.broker_version_data == BrokerVersionData((1, 0)) + def test_create_future_forwards_to_backend(self, net): + m = KafkaConnectionManager(net) + fut = m.create_future() + assert isinstance(fut, Future) + assert not fut.is_done + def test_client_dns_lookup_default(self, net): m = KafkaConnectionManager(net) assert m.config['client_dns_lookup'] == 'use_all_dns_ips' diff --git a/test/net/test_selector.py b/test/net/test_selector.py index 4f5707ee5..d2dbc9d81 100644 --- a/test/net/test_selector.py +++ b/test/net/test_selector.py @@ -742,6 +742,60 @@ def test_cancel_closes_ready_task(self): assert fired == [] +class TestCreateFuture: + """The pluggable-backend portability seam. + + Core coroutines call ``net.create_future()`` instead of constructing + ``Future`` directly so an alternate backend can return its own awaitable. + For ``NetworkSelector`` the native awaitable is ``kafka.future.Future``; + the contract these tests pin is: it is awaitable on the loop *and* honors + the callback-chain surface (``add_callback`` / ``add_errback`` / + ``add_both`` / ``success`` / ``failure``) that callers like + ``KafkaConnectionManager.wait_for`` rely on. + """ + + def test_returns_future(self): + net = NetworkSelector() + fut = net.create_future() + assert isinstance(fut, Future) + assert not fut.is_done + + def test_callback_chain(self): + net = NetworkSelector() + fut = net.create_future() + seen = [] + fut.add_callback(seen.append) + fut.success(42) + assert seen == [42] + assert fut.succeeded() and fut.value == 42 + + def test_errback_chain(self): + net = NetworkSelector() + fut = net.create_future() + errs = [] + fut.add_errback(errs.append) + exc = ValueError('boom') + fut.failure(exc) + assert errs == [exc] + assert fut.failed() + + def test_awaitable_on_loop(self): + net = NetworkSelector() + result = [] + + async def producer(fut): + fut.success('done') + + async def consumer(fut): + result.append(await fut) + + fut = net.create_future() + net.call_soon(consumer(fut)) + net.call_soon(producer(fut)) + net.drain() + assert result == ['done'] + + class TestSlowTaskMonitor: """Detection for tasks that hog the event loop (livelock guard). From cf23ff1101a14ddf5acf155f8a4a75bd69a14d5d Mon Sep 17 00:00:00 2001 From: Dana Powers Date: Tue, 30 Jun 2026 23:53:22 +0000 Subject: [PATCH 2/8] future: formalize the BackendFuture contract Define the contract that every pluggable backend's create_future() must satisfy, completing the create_future() seam from the previous commit. BackendFuture is a runtime_checkable Protocol documenting the surface core loop coroutines use (await + success/failure + the add_callback/errback/both callback core + is_done/value/exception) and, importantly, pinning the three semantic axes where backends could otherwise diverge: 1. Resolution thread: create_future() futures are created and resolved on the loop thread only; cross-thread handoffs use a plain thread-safe Future bridged via wait_for/run. 2. Fan-out: multiple awaiters and multiple callbacks are all resumed/invoked (a bare Twisted Deferred is single-consumer, so that backend wraps it per-awaiter). 3. Callback timing is unspecified (inline on the selector, deferred on asyncio); callers must not assume callbacks ran when success()/failure() returns. kafka.future.Future is the reference implementation and the base backends subclass, overriding only __await__ -- the single backend-specific hook. Add test/test_backend_future.py: a reusable, backend-agnostic conformance mixin (callback core, fan-out, await semantics) that Step 4's asyncio backend will reuse via an asyncio-driven subclass; wired up here for the selector. Co-Authored-By: Claude Opus 4.8 (1M context) --- kafka/future.py | 68 ++++++++++++++ test/test_backend_future.py | 178 ++++++++++++++++++++++++++++++++++++ 2 files changed, 246 insertions(+) create mode 100644 test/test_backend_future.py diff --git a/kafka/future.py b/kafka/future.py index ecaee6985..750a850bc 100644 --- a/kafka/future.py +++ b/kafka/future.py @@ -1,6 +1,7 @@ import functools import logging import threading +from typing import Any, Callable, Protocol, runtime_checkable from kafka.errors import RetriableError @@ -8,6 +9,18 @@ class Future: + """Callback/errback future, and the reference implementation of the + :class:`BackendFuture` contract. + + ``Future`` owns the portable callback core (``success`` / ``failure`` / + ``add_callback`` / ``add_errback`` / ``add_both`` / ``chain`` and the + ``is_done`` / ``value`` / ``exception`` state). A pluggable async backend + returns its own awaitable from ``net.create_future()`` by subclassing + ``Future`` and overriding only :meth:`__await__` — the single + backend-specific hook. The selector backend uses ``Future`` directly + (``__await__`` yields ``self``; the selector resumes the task on + resolution). See :class:`BackendFuture` for the full contract. + """ __slots__ = ('is_done', 'value', 'exception', '_callbacks', '_errbacks', '_lock') error_on_callbacks = False # and errbacks @@ -159,8 +172,63 @@ def chain(self, future): return self def __await__(self): + # The single backend-specific hook (see BackendFuture). For the + # selector backend the awaitable IS the Future: yield self and the + # selector re-enqueues the awaiting task when this future resolves. + # asyncio/Twisted backends override this to bridge to their native + # awaitable, leaving the callback core above untouched. if not self.is_done: yield self if self.exception: raise self.exception return self.value + + +@runtime_checkable +class BackendFuture(Protocol): + """Contract for the awaitable futures returned by ``net.create_future()``. + + A pluggable async backend (the kafka.net selector, asyncio, Twisted, ...) + returns its own future type from ``create_future()``. Core loop coroutines + touch it only through this surface, so the type is interchangeable across + backends. :class:`Future` is the reference implementation and the base + class backends subclass, overriding only ``__await__``. + + Pinned semantics — the three axes where backends could otherwise diverge: + + 1. **Resolution thread.** A future from ``create_future()`` is created and + resolved (``success`` / ``failure``) on the loop/IO thread only. + Cross-thread handoffs (a user thread blocking on a loop result) use a + plain thread-safe ``Future`` bridged via ``manager.wait_for`` / + ``manager.run`` — never a backend future awaited directly. Backends + whose native awaitable is loop-affine (``asyncio.Future``, Twisted + ``Deferred``) depend on this; their ``__await__`` adapter may assert it. + + 2. **Fan-out.** Multiple coroutines may ``await`` the same future and + multiple callbacks may be registered; all are resumed / invoked. (A bare + Twisted ``Deferred`` is single-consumer, so that backend wraps it + per-awaiter inside ``__await__``.) + + 3. **Callback timing is unspecified.** Callbacks may fire inline on the + resolving thread (selector) or on a later loop iteration (asyncio). + Callers must not assume callbacks have run when ``success`` / ``failure`` + returns. Code needing synchronous-resolution ordering uses a plain + ``Future`` (e.g. the producer batch latch), not a backend future. + + ``__await__`` is the only backend-specific member; the callback core is + portable and supplied by ``Future``. + """ + + is_done: bool + value: Any + exception: Any + + def __await__(self): ... + def success(self, value: Any) -> 'BackendFuture': ... + def failure(self, e: Any) -> 'BackendFuture': ... + def add_callback(self, f: Callable, *args: Any, **kwargs: Any) -> 'BackendFuture': ... + def add_errback(self, f: Callable, *args: Any, **kwargs: Any) -> 'BackendFuture': ... + 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: ... diff --git a/test/test_backend_future.py b/test/test_backend_future.py new file mode 100644 index 000000000..94bf502fe --- /dev/null +++ b/test/test_backend_future.py @@ -0,0 +1,178 @@ +"""Conformance suite for the BackendFuture contract (kafka/future.py). + +``BackendFutureContract`` is a reusable, backend-agnostic test mixin: it +pins the contract that every backend's ``create_future()`` must satisfy +(callback core, fan-out, await semantics). A backend provides two hooks -- +``make_future()`` (a fresh pending future) and ``drive(coros)`` (run the +given coroutines on the backend's loop to completion) -- and inherits the +whole suite. The selector implementation is below; Step 4's asyncio backend +reuses the same mixin with an asyncio-driven subclass. +""" +import pytest + +from kafka.future import Future, BackendFuture +from kafka.net.selector import NetworkSelector + + +class BackendFutureContract: + # --- hooks a backend must provide ------------------------------------- + def make_future(self): + raise NotImplementedError + + def drive(self, coros): + """Run coros on the backend loop until all complete.""" + raise NotImplementedError + + # --- typing / identity ------------------------------------------------ + def test_satisfies_protocol(self): + assert isinstance(self.make_future(), BackendFuture) + + # --- callback core (no loop needed) ----------------------------------- + def test_success_fires_callback(self): + fut = self.make_future() + seen = [] + fut.add_callback(seen.append) + fut.success(42) + assert seen == [42] + assert fut.is_done and fut.succeeded() + assert fut.value == 42 and fut.exception is None + + def test_failure_fires_errback(self): + fut = self.make_future() + errs = [] + fut.add_errback(errs.append) + exc = ValueError('boom') + fut.failure(exc) + assert errs == [exc] + assert fut.is_done and fut.failed() + assert fut.exception is exc + + def test_success_does_not_fire_errback(self): + fut = self.make_future() + cb, eb = [], [] + fut.add_callback(cb.append) + fut.add_errback(eb.append) + fut.success('ok') + assert cb == ['ok'] and eb == [] + + def test_add_both_on_success_and_failure(self): + ok = self.make_future() + bad = self.make_future() + seen_ok, seen_bad = [], [] + ok.add_both(seen_ok.append) + bad.add_both(seen_bad.append) + ok.success(1) + exc = ValueError('x') + bad.failure(exc) + assert seen_ok == [1] + assert seen_bad == [exc] + + def test_chain(self): + src = self.make_future() + dst = self.make_future() + src.chain(dst) + src.success('chained') + assert dst.succeeded() and dst.value == 'chained' + + def test_callback_added_after_resolution_fires_immediately(self): + fut = self.make_future() + fut.success(7) + seen = [] + fut.add_callback(seen.append) + assert seen == [7] + + # --- fan-out ---------------------------------------------------------- + def test_multiple_callbacks_all_fire(self): + fut = self.make_future() + a, b, c = [], [], [] + fut.add_callback(a.append) + fut.add_callback(b.append) + fut.add_callback(c.append) + fut.success('v') + assert a == b == c == ['v'] + + def test_multiple_awaiters_all_resume(self): + fut = self.make_future() + results = [] + + async def waiter(i): + results.append((i, await fut)) + + async def resolver(): + fut.success('v') + + self.drive([waiter(1), waiter(2), waiter(3), resolver()]) + assert sorted(results) == [(1, 'v'), (2, 'v'), (3, 'v')] + + # --- await semantics -------------------------------------------------- + def test_await_then_resolve(self): + fut = self.make_future() + results = [] + + async def waiter(): + results.append(await fut) + + async def resolver(): + fut.success('done') + + self.drive([waiter(), resolver()]) + assert results == ['done'] + + def test_resolve_then_await(self): + fut = self.make_future() + fut.success('early') + results = [] + + async def waiter(): + results.append(await fut) + + self.drive([waiter()]) + assert results == ['early'] + + def test_await_failure_raises(self): + fut = self.make_future() + caught = [] + + async def waiter(): + try: + await fut + except ValueError as exc: + caught.append(str(exc)) + + async def resolver(): + fut.failure(ValueError('nope')) + + self.drive([waiter(), resolver()]) + assert caught == ['nope'] + + +class TestNetworkSelectorBackendFuture(BackendFutureContract): + """The selector backend: create_future() returns a plain Future, driven + synchronously via NetworkSelector.drain() (no IO thread needed).""" + + @pytest.fixture(autouse=True) + def _net(self): + self.net = NetworkSelector() + try: + yield + finally: + self.net.close() + + def make_future(self): + return self.net.create_future() + + def drive(self, coros): + for coro in coros: + self.net.call_soon(coro) + self.net.drain() + + +class TestReferenceFutureIsBackendFuture: + """The bare kafka.future.Future (not via create_future) also conforms -- + it is the reference implementation of the contract.""" + + def test_isinstance(self): + assert isinstance(Future(), BackendFuture) + + def test_non_future_is_not_backend_future(self): + assert not isinstance(object(), BackendFuture) From 9f22d99c291e8498d8052cf92707d3d784b17b9f Mon Sep 17 00:00:00 2001 From: Dana Powers Date: Wed, 1 Jul 2026 00:23:18 +0000 Subject: [PATCH 3/8] future: complete the create_future() taxonomy (finish Step 1) Resolve the remaining direct Future() sites into a clean two-home model so there is no permanent third future category: - Loop-awaited results use create_future(). Convert the two pre-failed futures that were left plain but are consumed on the loop -- send()'s NodeNotReady early-return (sibling of send_request()'s create_future) and ensure_coordinator_ready_async's <0.8.2 NodeNotReady branch (consumed via wait_for). Drop the now-unused Future imports in manager/coordinator. - Everything else stays a plain thread-safe Future -- the eventual concurrent.futures.Future home -- because it is a cross-thread handoff or a fan-out lifecycle event, never awaited on the loop: * cluster request_update() _future and its pre-resolved set_topics/ add_topic siblings (user-thread-created handoff) * fetcher wait-for-first wakeup (user-thread-created handoff) * connection _close_future (cross-thread fan-out event; resolved on the loop on error/idle sweep or on a user thread via manager.close()) Label each remaining plain site inline with why it stays plain, and document the two-home rule on the BackendFuture contract. No behavior change (the converted sites are pre-resolved failures); full unit suite green. Co-Authored-By: Claude Opus 4.8 (1M context) --- kafka/cluster.py | 9 +++++++++ kafka/consumer/fetcher.py | 5 +++++ kafka/coordinator/base.py | 5 +++-- kafka/future.py | 5 +++++ kafka/net/connection.py | 6 +++++- kafka/net/manager.py | 5 +++-- 6 files changed, 30 insertions(+), 5 deletions(-) diff --git a/kafka/cluster.py b/kafka/cluster.py index 93fdf9bcb..73c4dc4ea 100644 --- a/kafka/cluster.py +++ b/kafka/cluster.py @@ -206,6 +206,8 @@ def set_topics(self, topics): for topic in topics: ensure_valid_topic_name(topic) if not set(topics).difference(self._topics): + # Pre-resolved sibling of request_update() (cross-thread handoff); + # already done, so it never yields -- safe as a plain Future. return Future().success(self) # TODO: handle future when old metadata request is currently in-flight # TODO: handle future when set_topics called multiple times before new request @@ -227,6 +229,8 @@ def add_topic(self, topic): """ ensure_valid_topic_name(topic) if topic in self._topics: + # Pre-resolved sibling of request_update() (cross-thread handoff); + # already done, so it never yields -- safe as a plain Future. return Future().success(self) # TODO: handle future when old metadata request is currently in-flight self._topics.add(topic) @@ -430,6 +434,11 @@ def request_update(self): with self._lock: self._need_update = True if not self._future or self._future.is_done: + # Cross-thread handoff: created here on a user thread (callers + # invoke request_update() off-loop), resolved on the loop, and + # awaited only via manager.wait_for's wrapper -- never directly. + # Stays a plain thread-safe Future -> concurrent.futures.Future + # candidate, not create_future(). self._future = Future() ret = self._future if self._manager: diff --git a/kafka/consumer/fetcher.py b/kafka/consumer/fetcher.py index e89873b47..a19dfde35 100644 --- a/kafka/consumer/fetcher.py +++ b/kafka/consumer/fetcher.py @@ -250,6 +250,11 @@ def fetch_records(self, max_records=None, update_offsets=True, timeout_ms=None): if not waited_on: return records, True # nothing pending; caller should sleep + # Cross-thread wait-for-first: created here on the user thread, resolved + # on the loop via the waited_on callbacks, and bridged to the user + # thread via self._net.run(wait_for, ...) -- never awaited directly. + # Stays a plain thread-safe Future -> concurrent.futures.Future + # candidate, not create_future(). wakeup = Future() def _wake(_): if not wakeup.is_done: diff --git a/kafka/coordinator/base.py b/kafka/coordinator/base.py index 1ceff5634..8016e3e67 100644 --- a/kafka/coordinator/base.py +++ b/kafka/coordinator/base.py @@ -7,7 +7,6 @@ from kafka.coordinator.heartbeat import Heartbeat from kafka import errors as Errors -from kafka.future import Future from kafka.metrics import AnonMeasurable from kafka.metrics.stats import Avg, Count, Max, Rate from kafka.net.wakeup_notifier import WakeupNotifier @@ -352,7 +351,9 @@ async def ensure_coordinator_ready_async(self, timeout_ms=None): if self.config['api_version'] < (0, 8, 2): maybe_coordinator_id = self._client.least_loaded_node() if maybe_coordinator_id is None: - future = Future().failure(Errors.NodeNotReadyError('coordinator')) + # Pre-failed sibling of lookup_coordinator(); consumed via + # wait_for on the loop, so mint it from the backend too. + future = self._manager.create_future().failure(Errors.NodeNotReadyError('coordinator')) else: self.coordinator_id = maybe_coordinator_id return not timer.expired diff --git a/kafka/future.py b/kafka/future.py index 750a850bc..2fc9af9f1 100644 --- a/kafka/future.py +++ b/kafka/future.py @@ -204,6 +204,11 @@ class backends subclass, overriding only ``__await__``. whose native awaitable is loop-affine (``asyncio.Future``, Twisted ``Deferred``) depend on this; their ``__await__`` adapter may assert it. + There is no permanent third future category: a call site uses + ``create_future()`` when a coroutine awaits the result on the loop, and + a plain ``Future`` otherwise (a cross-thread handoff or a fan-out + lifecycle event — the eventual ``concurrent.futures.Future`` home). + 2. **Fan-out.** Multiple coroutines may ``await`` the same future and multiple callbacks may be registered; all are resumed / invoked. (A bare Twisted ``Deferred`` is single-consumer, so that backend wraps it diff --git a/kafka/net/connection.py b/kafka/net/connection.py index 5b1a67377..776ffdc1f 100644 --- a/kafka/net/connection.py +++ b/kafka/net/connection.py @@ -54,7 +54,11 @@ def __init__(self, net, node_id=None, broker_version_data=None, **configs): self.connected = False self.initializing = True self._init_future = net.create_future() # awaited via __await__; backend-native - self._close_future = Future() # callback-only, never awaited + # Cross-thread fan-out event (resolved on the loop on error/idle sweep, + # or on a user thread via manager.close()); never awaited, only + # callbacks. Stays a plain thread-safe Future -> concurrent.futures.Future + # candidate, not create_future(). + self._close_future = Future() self.in_flight_requests = collections.deque() self.broker_version_data = broker_version_data self._api_versions_idx = ApiVersionsRequest.max_version # version of ApiVersionsRequest to try on first connect diff --git a/kafka/net/manager.py b/kafka/net/manager.py index ab034a4d3..204f97314 100644 --- a/kafka/net/manager.py +++ b/kafka/net/manager.py @@ -13,7 +13,6 @@ import kafka.errors as Errors from kafka.net.wakeup_notifier import WakeupNotifier from kafka.protocol.broker_version_data import BrokerVersionData -from kafka.future import Future from kafka.version import __version__ @@ -299,7 +298,9 @@ def send(self, request, node_id=None, request_timeout_ms=None): try: conn = self.get_connection(node_id) except Errors.NodeNotReadyError as e: - return Future().failure(e) + # Pre-failed sibling of send_request()'s create_future(); awaited + # on the loop by the same caller, so mint it from the backend too. + return self.create_future().failure(e) else: return conn.send_request(request, request_timeout_ms=request_timeout_ms) From 43cd39dc8b4665af066c125ddf597d9c8d65ef19 Mon Sep 17 00:00:00 2001 From: Dana Powers Date: Tue, 7 Jul 2026 12:49:30 -0700 Subject: [PATCH 4/8] sans unicode --- kafka/future.py | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/kafka/future.py b/kafka/future.py index 2fc9af9f1..c1419521b 100644 --- a/kafka/future.py +++ b/kafka/future.py @@ -16,7 +16,7 @@ class Future: ``add_callback`` / ``add_errback`` / ``add_both`` / ``chain`` and the ``is_done`` / ``value`` / ``exception`` state). A pluggable async backend returns its own awaitable from ``net.create_future()`` by subclassing - ``Future`` and overriding only :meth:`__await__` — the single + ``Future`` and overriding only :meth:`__await__` -- the single backend-specific hook. The selector backend uses ``Future`` directly (``__await__`` yields ``self``; the selector resumes the task on resolution). See :class:`BackendFuture` for the full contract. @@ -194,20 +194,20 @@ class BackendFuture(Protocol): backends. :class:`Future` is the reference implementation and the base class backends subclass, overriding only ``__await__``. - Pinned semantics — the three axes where backends could otherwise diverge: + Pinned semantics -- the three axes where backends could otherwise diverge: 1. **Resolution thread.** A future from ``create_future()`` is created and resolved (``success`` / ``failure``) on the loop/IO thread only. Cross-thread handoffs (a user thread blocking on a loop result) use a plain thread-safe ``Future`` bridged via ``manager.wait_for`` / - ``manager.run`` — never a backend future awaited directly. Backends + ``manager.run`` -- never a backend future awaited directly. Backends whose native awaitable is loop-affine (``asyncio.Future``, Twisted ``Deferred``) depend on this; their ``__await__`` adapter may assert it. There is no permanent third future category: a call site uses ``create_future()`` when a coroutine awaits the result on the loop, and a plain ``Future`` otherwise (a cross-thread handoff or a fan-out - lifecycle event — the eventual ``concurrent.futures.Future`` home). + lifecycle event -- the eventual ``concurrent.futures.Future`` home). 2. **Fan-out.** Multiple coroutines may ``await`` the same future and multiple callbacks may be registered; all are resumed / invoked. (A bare From 865d1ecf103fc02130d9ba660abd2d3e869b5248 Mon Sep 17 00:00:00 2001 From: Dana Powers Date: Tue, 7 Jul 2026 14:27:58 -0700 Subject: [PATCH 5/8] kafka/net/manager.py: wait_for now always awaits a create_future() wrapper instead of the passed future directly -- a plain Future isn't awaitable on every backend (asyncio rejects a bare `yield self`); behavior-preserving on the selector. --- kafka/net/manager.py | 21 +++++++++++++-------- 1 file changed, 13 insertions(+), 8 deletions(-) diff --git a/kafka/net/manager.py b/kafka/net/manager.py index 204f97314..55501a2ea 100644 --- a/kafka/net/manager.py +++ b/kafka/net/manager.py @@ -411,8 +411,10 @@ async def wait_for(self, future, timeout_ms): future is not cancelled on timeout - it continues to run; the timeout only unblocks the awaiter. """ - if timeout_ms is None: - return await future + # Always await a backend-native wrapper, never `future` directly: + # `future` may be a plain thread-safe Future which isn't awaitable on + # every backend (e.g. asyncio rejects a bare `yield self`). We touch it + # only via callbacks. (create_future() gives the backend's awaitable.) wrapper = self._net.create_future() def _on_success(value): if not wrapper.is_done: @@ -422,15 +424,18 @@ def _on_failure(exc): wrapper.failure(exc) future.add_callback(_on_success) future.add_errback(_on_failure) - def _on_timeout(): - if not wrapper.is_done: - wrapper.failure(Errors.KafkaTimeoutError( - 'Timed out after %s ms' % timeout_ms)) - timer = self._net.call_later(timeout_ms / 1000, _on_timeout) + timer = None + if timeout_ms is not None: + def _on_timeout(): + if not wrapper.is_done: + wrapper.failure(Errors.KafkaTimeoutError( + 'Timed out after %s ms' % timeout_ms)) + timer = self._net.call_later(timeout_ms / 1000, _on_timeout) try: return await wrapper finally: - self._net.cancel(timer) + if timer is not None: + self._net.cancel(timer) def create_future(self): """Create a Future suitable for awaiting on the underlying loop. From c008150483449349ba693044b8e877ad6b42a9ad Mon Sep 17 00:00:00 2001 From: Dana Powers Date: Tue, 7 Jul 2026 21:53:27 +0000 Subject: [PATCH 6/8] future: split awaitable SelectorFuture from thread-safe Future kafka.future.Future was doing double duty: the selector's loop-awaitable future (create_future()) AND the thread-safe cross-thread handoff future (FutureProduceResult, fetcher wakeup, cluster update future, ...). __await__ on the base made every Future look loop-awaitable, so awaiting a handoff future worked by accident on the selector and would only break later on another backend. Separate the two roles so the distinction is type-enforced: - kafka.future.Future drops __await__ -- it is now purely the thread-safe callback/handoff primitive (the eventual concurrent.futures.Future role). - kafka.net.selector.SelectorFuture (a Future subclass) adds __await__ (yield self); NetworkSelector.create_future() and call_soon_with_future() return it. Every future a loop coroutine awaits is a SelectorFuture. - The BackendFuture contract moves from kafka.future to a new kafka.net.backend module (it is a net/backend concept). SelectorFuture is its reference implementation; a plain Future is deliberately NOT a BackendFuture (no __await__). Now awaiting a plain handoff Future raises immediately (no __await__) instead of silently working on one backend and breaking on another. Removing __await__ was self-verifying: the only breakages were test mocks that returned a plain Future() where production returns create_future() (connection/sasl init _send_request stubs, fetcher/coordinator send stubs, orphan-cycle regression). Those tests now obtain an awaitable future via the backend factory (net/manager/conn .create_future()) rather than constructing SelectorFuture directly. No production code awaited a plain Future. Full unit suite green. Co-Authored-By: Claude Opus 4.8 (1M context) --- kafka/future.py | 86 +++----------------------- kafka/net/backend.py | 66 ++++++++++++++++++++ kafka/net/selector.py | 28 +++++++-- test/consumer/test_coordinator.py | 4 +- test/consumer/test_fetcher.py | 6 +- test/net/test_connection.py | 10 +-- test/net/test_manager.py | 7 ++- test/net/test_sasl_reauthentication.py | 2 +- test/net/test_selector.py | 14 ++--- test/test_backend_future.py | 19 +++--- test/test_future.py | 40 +++++------- 11 files changed, 149 insertions(+), 133 deletions(-) create mode 100644 kafka/net/backend.py diff --git a/kafka/future.py b/kafka/future.py index c1419521b..5c94be734 100644 --- a/kafka/future.py +++ b/kafka/future.py @@ -1,7 +1,6 @@ import functools import logging import threading -from typing import Any, Callable, Protocol, runtime_checkable from kafka.errors import RetriableError @@ -9,17 +8,19 @@ class Future: - """Callback/errback future, and the reference implementation of the - :class:`BackendFuture` contract. + """Thread-safe callback/errback future -- a cross-thread handoff primitive. ``Future`` owns the portable callback core (``success`` / ``failure`` / ``add_callback`` / ``add_errback`` / ``add_both`` / ``chain`` and the - ``is_done`` / ``value`` / ``exception`` state). A pluggable async backend - returns its own awaitable from ``net.create_future()`` by subclassing - ``Future`` and overriding only :meth:`__await__` -- the single - backend-specific hook. The selector backend uses ``Future`` directly - (``__await__`` yields ``self``; the selector resumes the task on - resolution). See :class:`BackendFuture` for the full contract. + ``is_done`` / ``value`` / ``exception`` state) and is safe to resolve from + any thread. It is deliberately **not** awaitable: awaiting happens only on + the event loop, via the backend's loop-awaitable future from + ``net.create_future()`` (``kafka.net.selector.SelectorFuture`` for the + selector), which subclasses ``Future`` and adds ``__await__``. Keeping + ``__await__`` off the base makes the invariant type-enforced -- awaiting a + plain handoff ``Future`` raises immediately rather than silently working on + one backend and breaking on another. See ``kafka.net.backend.BackendFuture`` + for the awaitable contract. """ __slots__ = ('is_done', 'value', 'exception', '_callbacks', '_errbacks', '_lock') error_on_callbacks = False # and errbacks @@ -170,70 +171,3 @@ def add_both(self, f, *args, **kwargs): def chain(self, future): self._add_cb_eb(future.success, future.failure) return self - - def __await__(self): - # The single backend-specific hook (see BackendFuture). For the - # selector backend the awaitable IS the Future: yield self and the - # selector re-enqueues the awaiting task when this future resolves. - # asyncio/Twisted backends override this to bridge to their native - # awaitable, leaving the callback core above untouched. - if not self.is_done: - yield self - if self.exception: - raise self.exception - return self.value - - -@runtime_checkable -class BackendFuture(Protocol): - """Contract for the awaitable futures returned by ``net.create_future()``. - - A pluggable async backend (the kafka.net selector, asyncio, Twisted, ...) - returns its own future type from ``create_future()``. Core loop coroutines - touch it only through this surface, so the type is interchangeable across - backends. :class:`Future` is the reference implementation and the base - class backends subclass, overriding only ``__await__``. - - Pinned semantics -- the three axes where backends could otherwise diverge: - - 1. **Resolution thread.** A future from ``create_future()`` is created and - resolved (``success`` / ``failure``) on the loop/IO thread only. - Cross-thread handoffs (a user thread blocking on a loop result) use a - plain thread-safe ``Future`` bridged via ``manager.wait_for`` / - ``manager.run`` -- never a backend future awaited directly. Backends - whose native awaitable is loop-affine (``asyncio.Future``, Twisted - ``Deferred``) depend on this; their ``__await__`` adapter may assert it. - - There is no permanent third future category: a call site uses - ``create_future()`` when a coroutine awaits the result on the loop, and - a plain ``Future`` otherwise (a cross-thread handoff or a fan-out - lifecycle event -- the eventual ``concurrent.futures.Future`` home). - - 2. **Fan-out.** Multiple coroutines may ``await`` the same future and - multiple callbacks may be registered; all are resumed / invoked. (A bare - Twisted ``Deferred`` is single-consumer, so that backend wraps it - per-awaiter inside ``__await__``.) - - 3. **Callback timing is unspecified.** Callbacks may fire inline on the - resolving thread (selector) or on a later loop iteration (asyncio). - Callers must not assume callbacks have run when ``success`` / ``failure`` - returns. Code needing synchronous-resolution ordering uses a plain - ``Future`` (e.g. the producer batch latch), not a backend future. - - ``__await__`` is the only backend-specific member; the callback core is - portable and supplied by ``Future``. - """ - - is_done: bool - value: Any - exception: Any - - def __await__(self): ... - def success(self, value: Any) -> 'BackendFuture': ... - def failure(self, e: Any) -> 'BackendFuture': ... - def add_callback(self, f: Callable, *args: Any, **kwargs: Any) -> 'BackendFuture': ... - def add_errback(self, f: Callable, *args: Any, **kwargs: Any) -> 'BackendFuture': ... - 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: ... diff --git a/kafka/net/backend.py b/kafka/net/backend.py new file mode 100644 index 000000000..075818c13 --- /dev/null +++ b/kafka/net/backend.py @@ -0,0 +1,66 @@ +"""Pluggable async-backend contracts. + +For now this holds the :class:`BackendFuture` contract -- 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. +""" +from typing import Any, Callable, Protocol, runtime_checkable + + +@runtime_checkable +class BackendFuture(Protocol): + """Contract for the awaitable futures returned by ``net.create_future()``. + + A pluggable async backend (the kafka.net selector, asyncio, Twisted, ...) + returns its own future type from ``create_future()``. Core loop coroutines + touch it only through this surface, so the type is interchangeable across + backends. The selector's ``SelectorFuture`` is the reference implementation: + it subclasses the thread-safe ``kafka.future.Future`` (the portable callback + core) and adds ``__await__``. A plain ``Future`` is deliberately NOT a + BackendFuture -- it has no ``__await__`` -- so awaiting a cross-thread + handoff future fails loudly instead of silently working on one backend. + + Pinned semantics -- the three axes where backends could otherwise diverge: + + 1. **Resolution thread.** A future from ``create_future()`` is created and + resolved (``success`` / ``failure``) on the loop/IO thread only. + Cross-thread handoffs (a user thread blocking on a loop result) use a + plain thread-safe ``Future`` bridged via ``manager.wait_for`` / + ``manager.run`` -- never a backend future awaited directly. Backends + whose native awaitable is loop-affine (``asyncio.Future``, Twisted + ``Deferred``) depend on this; their ``__await__`` adapter may assert it. + + There is no permanent third future category: a call site uses + ``create_future()`` when a coroutine awaits the result on the loop, and + a plain ``Future`` otherwise (a cross-thread handoff or a fan-out + lifecycle event -- the eventual ``concurrent.futures.Future`` home). + + 2. **Fan-out.** Multiple coroutines may ``await`` the same future and + multiple callbacks may be registered; all are resumed / invoked. (A bare + Twisted ``Deferred`` is single-consumer, so that backend wraps it + per-awaiter inside ``__await__``.) + + 3. **Callback timing is unspecified.** Callbacks may fire inline on the + resolving thread (selector) or on a later loop iteration (asyncio). + Callers must not assume callbacks have run when ``success`` / ``failure`` + returns. Code needing synchronous-resolution ordering uses a plain + ``Future`` (e.g. the producer batch latch), not a backend future. + + ``__await__`` is the only backend-specific member; the callback core is + portable and supplied by ``Future``. + """ + + is_done: bool + value: Any + exception: Any + + def __await__(self): ... + def success(self, value: Any) -> 'BackendFuture': ... + def failure(self, e: Any) -> 'BackendFuture': ... + def add_callback(self, f: Callable, *args: Any, **kwargs: Any) -> 'BackendFuture': ... + def add_errback(self, f: Callable, *args: Any, **kwargs: Any) -> 'BackendFuture': ... + 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: ... diff --git a/kafka/net/selector.py b/kafka/net/selector.py index 384d93322..7726a37b9 100644 --- a/kafka/net/selector.py +++ b/kafka/net/selector.py @@ -40,6 +40,26 @@ def _initialize_coro(maybe_coro): raise TypeError('Generator or coroutine not found: %s' % type(maybe_coro)) +class SelectorFuture(Future): + """The NetworkSelector's loop-awaitable future (see backend.BackendFuture). + + ``kafka.future.Future`` is the thread-safe callback/handoff core with no + ``__await__``; ``SelectorFuture`` adds it: ``yield self`` suspends the + awaiting coroutine, and the selector re-enqueues it when the future + resolves (Task/_poll_once dispatch on ``isinstance(_, Future)``, which this + subclass satisfies). Returned by ``NetworkSelector.create_future()`` and + used for every future a loop coroutine awaits. + """ + __slots__ = () + + def __await__(self): + if not self.is_done: + yield self + if self.exception: + raise self.exception + return self.value + + class KernelEvent: def __init__(self, method, *args): self.method = method @@ -383,7 +403,7 @@ def call_soon_with_future(self, coro, *args): if hasattr(coro, '__await__'): if args: raise ValueError('initiated coroutine does not accept args') - future = Future() + future = SelectorFuture() # returned to callers who await it async def wrapper(): try: future.success(await self._invoke(coro, *args)) @@ -446,14 +466,14 @@ def reschedule(self, when, task): return task def create_future(self): - """Create a Future suitable for awaiting on this backend's loop. + """Create a loop-awaitable future (see backend.BackendFuture). Portability seam for pluggable backends: core coroutines call this instead of constructing ``Future`` directly, so an alternate backend (asyncio, Twisted) can return its own awaitable type. The selector's - native awaitable is ``kafka.future.Future``. + native awaitable is ``SelectorFuture`` (a ``Future`` with ``__await__``). """ - return Future() + return SelectorFuture() def sleep(self, delay): return KernelEvent('_sleep', delay) diff --git a/test/consumer/test_coordinator.py b/test/consumer/test_coordinator.py index 2fc3ed431..bb6a2c4c8 100644 --- a/test/consumer/test_coordinator.py +++ b/test/consumer/test_coordinator.py @@ -1370,7 +1370,7 @@ def test_join_group_async_returns_false_on_short_timeout_and_caches_task( seeded_coord.state = MemberState.UNJOINED # JoinGroup response future controlled by the test. Hangs until released. - join_response_pending = Future() + join_response_pending = seeded_coord._manager.create_future() # awaited by slow_join_handler join_request_count = [0] async def slow_join_handler(api_key, api_version, correlation_id, request_bytes): @@ -1489,7 +1489,7 @@ def test_heartbeat(mocker, coordinator): # spin); using side_effect with an async function that awaits the Future # forces the suspension we want. The Mock's call_count then verifies the # loop fired exactly once. - blocked_send = Future() + blocked_send = coordinator._manager.create_future() async def _hang(*args, **kwargs): await blocked_send mocker.patch.object(coordinator, '_send_heartbeat_request', side_effect=_hang) diff --git a/test/consumer/test_fetcher.py b/test/consumer/test_fetcher.py index 8a3878b5d..03b09aaff 100644 --- a/test/consumer/test_fetcher.py +++ b/test/consumer/test_fetcher.py @@ -316,7 +316,7 @@ def test__send_list_offsets_requests(fetcher, manager, net, mocker): pending = [] async def fake_send(node_id, timestamps): - f = Future() + f = fetcher._manager.create_future() # awaited below pending.append(f) return await f mocked_send = mocker.patch.object(fetcher, "_send_list_offsets_request", side_effect=fake_send) @@ -370,7 +370,7 @@ def test__send_list_offsets_requests_multiple_nodes(fetcher, manager, net, mocke send_futures = [] async def fake_send(node_id, timestamps): - f = Future() + f = fetcher._manager.create_future() # awaited below send_futures.append((node_id, timestamps, f)) return await f mocked_send = mocker.patch.object(fetcher, "_send_list_offsets_request", side_effect=fake_send) @@ -1268,7 +1268,7 @@ def test_timeout_raises(self, fetcher, mocker): # Awaits a future that never completes async def fake_send(ts): - await Future() + await fetcher._manager.create_future() mocker.patch.object(fetcher, '_send_list_offsets_requests', side_effect=fake_send) with pytest.raises(Errors.KafkaTimeoutError): diff --git a/test/net/test_connection.py b/test/net/test_connection.py index 04a4f3d51..005f9ae3b 100644 --- a/test/net/test_connection.py +++ b/test/net/test_connection.py @@ -92,7 +92,9 @@ def _run_initialize(self, net, conn, responses): original_send = conn._send_request def mock_send_request(request, **kwargs): requests_sent.append(request.API_VERSION) - f = Future() + # Match production _send_request: a loop-awaitable SelectorFuture, + # not a plain (non-awaitable) Future -- the conn awaits this. + f = net.create_future() try: resp = next(response_iter) except StopIteration: @@ -531,7 +533,7 @@ def test_sasl_authenticate_success(self, net): responses = iter([handshake_response, auth_response]) def mock_send_request(request, **kwargs): - f = Future() + f = conn.net.create_future() # loop-awaitable, like production f.success(next(responses)) return f conn._send_request = mock_send_request @@ -569,7 +571,7 @@ def test_sasl_authenticate_auth_failure(self, net): responses = iter([handshake_response, auth_response]) def mock_send_request(request, **kwargs): - f = Future() + f = conn.net.create_future() # loop-awaitable, like production f.success(next(responses)) return f conn._send_request = mock_send_request @@ -591,7 +593,7 @@ def _drive_handshake_with_recording_mechanism(self, net, conn): auth_response.session_lifetime_ms = 0 responses = iter([handshake_response, auth_response]) def mock_send_request(request, **kwargs): - f = Future() + f = conn.net.create_future() # loop-awaitable, like production f.success(next(responses)) return f conn._send_request = mock_send_request diff --git a/test/net/test_manager.py b/test/net/test_manager.py index 5d767c367..036b6fa6a 100644 --- a/test/net/test_manager.py +++ b/test/net/test_manager.py @@ -514,9 +514,10 @@ def aggressive_poll_once(self, timeout=None): monkeypatch.setattr(NetworkSelector, '_poll_once', aggressive_poll_once) async def hangs_then_times_out(): - # Awaits a bare Future that nothing references externally -- - # exactly the orphan-cycle shape that CPython's gc collects. - await Future() + # Awaits a bare (loop-awaitable) future that nothing references + # externally -- exactly the orphan-cycle shape that CPython's gc + # collects. + await manager.create_future() # wait_for should fail with KafkaTimeoutError, not GeneratorExit. async def waiter(): diff --git a/test/net/test_sasl_reauthentication.py b/test/net/test_sasl_reauthentication.py index c238063fa..fe6ba4612 100644 --- a/test/net/test_sasl_reauthentication.py +++ b/test/net/test_sasl_reauthentication.py @@ -214,7 +214,7 @@ def test_session_lifetime_captured_from_v1_response(self, net): responses = iter([handshake_response, auth_response]) def mock_send_request(request, **kwargs): - f = Future() + f = conn.net.create_future() # loop-awaitable, like production f.success(next(responses)) return f conn._send_request = mock_send_request diff --git a/test/net/test_selector.py b/test/net/test_selector.py index d2dbc9d81..e4fbbf886 100644 --- a/test/net/test_selector.py +++ b/test/net/test_selector.py @@ -571,7 +571,7 @@ def task(): def test_await_future_already_done(self): net = NetworkSelector() - f = Future() + f = net.create_future() f.success(99) results = [] @@ -584,7 +584,7 @@ async def waiter(): def test_await_future_already_failed(self): net = NetworkSelector() - f = Future() + f = net.create_future() f.failure(ValueError('oops')) errors = [] @@ -648,7 +648,7 @@ def test_close(self): def test_await_future_resolved(self): net = NetworkSelector() - f = Future() + f = net.create_future() f.success(42) results = [] @@ -661,7 +661,7 @@ async def task(): def test_await_future_pending(self): net = NetworkSelector() - f = Future() + f = net.create_future() results = [] done = Future() @@ -681,7 +681,7 @@ async def resolver(): def test_await_future_failure(self): net = NetworkSelector() - f = Future() + f = net.create_future() errors = [] done = Future() @@ -703,8 +703,8 @@ async def resolver(): def test_await_multiple_futures(self): net = NetworkSelector() - f1 = Future() - f2 = Future() + f1 = net.create_future() + f2 = net.create_future() results = [] done = Future() diff --git a/test/test_backend_future.py b/test/test_backend_future.py index 94bf502fe..b38e16e05 100644 --- a/test/test_backend_future.py +++ b/test/test_backend_future.py @@ -10,7 +10,8 @@ """ import pytest -from kafka.future import Future, BackendFuture +from kafka.future import Future +from kafka.net.backend import BackendFuture from kafka.net.selector import NetworkSelector @@ -147,7 +148,7 @@ async def resolver(): class TestNetworkSelectorBackendFuture(BackendFutureContract): - """The selector backend: create_future() returns a plain Future, driven + """The selector backend: create_future() returns a SelectorFuture, driven synchronously via NetworkSelector.drain() (no IO thread needed).""" @pytest.fixture(autouse=True) @@ -167,12 +168,16 @@ def drive(self, coros): self.net.drain() -class TestReferenceFutureIsBackendFuture: - """The bare kafka.future.Future (not via create_future) also conforms -- - it is the reference implementation of the contract.""" +class TestBackendFutureTypeEnforcement: + """A backend's create_future() result satisfies BackendFuture -- covered by + TestNetworkSelectorBackendFuture.test_satisfies_protocol above. Here we pin + the negative: the plain thread-safe Future (a cross-thread handoff) + deliberately does NOT (no __await__), which turns 'awaited a handoff future' + into a loud error rather than a silent backend-specific bug.""" - def test_isinstance(self): - assert isinstance(Future(), BackendFuture) + def test_plain_future_is_not_backend_future(self): + assert not isinstance(Future(), BackendFuture) + assert not hasattr(Future(), '__await__') def test_non_future_is_not_backend_future(self): assert not isinstance(object(), BackendFuture) diff --git a/test/test_future.py b/test/test_future.py index 5b0b5b9e2..d87e4c906 100644 --- a/test/test_future.py +++ b/test/test_future.py @@ -1,31 +1,8 @@ -import pytest - from kafka.future import Future -class TestFutureAwait: - def test_await_resolved_success(self): - f = Future() - f.success(42) - coro = f.__await__() - with pytest.raises(StopIteration) as exc_info: - next(coro) - assert exc_info.value.value == 42 - - def test_await_resolved_failure(self): - f = Future() - f.failure(ValueError('bad')) - coro = f.__await__() - with pytest.raises(ValueError, match='bad'): - next(coro) - - def test_await_pending_yields_self(self): - f = Future() - coro = f.__await__() - yielded = next(coro) - assert yielded is f - - def test_callbacks_still_work(self): +class TestFutureCallbacks: + def test_callbacks(self): f = Future() results = [] f.add_callback(lambda v: results.append(('cb', v))) @@ -33,7 +10,7 @@ def test_callbacks_still_work(self): f.success(42) assert results == [('cb', 42)] - def test_errbacks_still_work(self): + def test_errbacks(self): f = Future() results = [] f.add_errback(lambda e: results.append(('eb', str(e)))) @@ -55,3 +32,14 @@ def test_chain_failure(self): f1.failure(ValueError('err')) assert f2.failed() assert isinstance(f2.exception, ValueError) + + +class TestFutureNotAwaitable: + """The plain Future is a thread-safe handoff primitive, not loop-awaitable. + __await__ lives on the backend future (e.g. SelectorFuture from + create_future()), not the base, so awaiting a handoff Future fails loudly. + The backend futures' await behavior is covered by test_selector.py and the + BackendFuture conformance suite.""" + + def test_no_dunder_await(self): + assert not hasattr(Future(), '__await__') From 0f5bf93195bd23e55e1495c71336e13c51e4723b Mon Sep 17 00:00:00 2001 From: Dana Powers Date: Fri, 10 Jul 2026 14:16:02 -0700 Subject: [PATCH 7/8] mv test_backend_future -> test/net/ --- test/{ => net}/test_backend_future.py | 0 1 file changed, 0 insertions(+), 0 deletions(-) rename test/{ => net}/test_backend_future.py (100%) diff --git a/test/test_backend_future.py b/test/net/test_backend_future.py similarity index 100% rename from test/test_backend_future.py rename to test/net/test_backend_future.py From f8a679d6330555f3576f3d62f7ecbacd3c12c60d Mon Sep 17 00:00:00 2001 From: Dana Powers Date: Fri, 10 Jul 2026 14:40:16 -0700 Subject: [PATCH 8/8] comment / docstring updates --- kafka/cluster.py | 33 ++++++++++++++++++++++----------- kafka/consumer/fetcher.py | 7 +------ kafka/net/backend.py | 5 ----- kafka/net/connection.py | 4 +--- 4 files changed, 24 insertions(+), 25 deletions(-) diff --git a/kafka/cluster.py b/kafka/cluster.py index 73c4dc4ea..ad62c2f2f 100644 --- a/kafka/cluster.py +++ b/kafka/cluster.py @@ -423,23 +423,34 @@ def refresh_backoff(self): return self.config['retry_backoff_ms'] def request_update(self): - """Flags metadata for update, return Future() - - Actual update must be handled separately. This method will only - change the reported ttl() + """Trigger a metadata refresh; return a completion token (not awaitable). + + This is a cross-thread trigger, not a coroutine: it flags metadata as + stale (changing the reported ttl()), wakes the refresh loop, and returns + a token Future that resolves when the next update lands. It is safe to + call from any thread -- including user threads off the IO loop -- which + is precisely why the returned Future is a plain thread-safe handoff and + NOT a backend awaitable: a loop-affine future (create_future()) can't be + minted off the loop thread. + + Do not ``await`` the returned Future directly. Await it at the edge via + ``manager.wait_for(future, timeout_ms)``, which resolves it through the + backend's own awaitable: + on-loop: await self._manager.wait_for(cluster.request_update(), t) + off-loop: self._net.run(self._manager.wait_for, cluster.request_update(), None) + Many callers want only the flag+wake side effect and discard the token. + + On-loop callers that simply want to await a refresh can instead use the + awaitable sibling ``await refresh_metadata()`` and skip the token. Returns: - kafka.future.Future (value will be the cluster object after update) + kafka.future.Future: completion token, resolved when the next + metadata update (or failure) is applied. """ with self._lock: self._need_update = True if not self._future or self._future.is_done: - # Cross-thread handoff: created here on a user thread (callers - # invoke request_update() off-loop), resolved on the loop, and - # awaited only via manager.wait_for's wrapper -- never directly. - # Stays a plain thread-safe Future -> concurrent.futures.Future - # candidate, not create_future(). - self._future = Future() + self._future = Future() # simple future; not backend / no await ret = self._future if self._manager: self.start_refresh_loop() diff --git a/kafka/consumer/fetcher.py b/kafka/consumer/fetcher.py index a19dfde35..14a0f9ba4 100644 --- a/kafka/consumer/fetcher.py +++ b/kafka/consumer/fetcher.py @@ -250,12 +250,7 @@ def fetch_records(self, max_records=None, update_offsets=True, timeout_ms=None): if not waited_on: return records, True # nothing pending; caller should sleep - # Cross-thread wait-for-first: created here on the user thread, resolved - # on the loop via the waited_on callbacks, and bridged to the user - # thread via self._net.run(wait_for, ...) -- never awaited directly. - # Stays a plain thread-safe Future -> concurrent.futures.Future - # candidate, not create_future(). - wakeup = Future() + wakeup = Future() # plain future, not backend / no await def _wake(_): if not wakeup.is_done: wakeup.success(None) diff --git a/kafka/net/backend.py b/kafka/net/backend.py index 075818c13..58cba7f2c 100644 --- a/kafka/net/backend.py +++ b/kafka/net/backend.py @@ -31,11 +31,6 @@ class BackendFuture(Protocol): whose native awaitable is loop-affine (``asyncio.Future``, Twisted ``Deferred``) depend on this; their ``__await__`` adapter may assert it. - There is no permanent third future category: a call site uses - ``create_future()`` when a coroutine awaits the result on the loop, and - a plain ``Future`` otherwise (a cross-thread handoff or a fan-out - lifecycle event -- the eventual ``concurrent.futures.Future`` home). - 2. **Fan-out.** Multiple coroutines may ``await`` the same future and multiple callbacks may be registered; all are resumed / invoked. (A bare Twisted ``Deferred`` is single-consumer, so that backend wraps it diff --git a/kafka/net/connection.py b/kafka/net/connection.py index 776ffdc1f..45ca27f8d 100644 --- a/kafka/net/connection.py +++ b/kafka/net/connection.py @@ -55,9 +55,7 @@ def __init__(self, net, node_id=None, broker_version_data=None, **configs): self.initializing = True self._init_future = net.create_future() # awaited via __await__; backend-native # Cross-thread fan-out event (resolved on the loop on error/idle sweep, - # or on a user thread via manager.close()); never awaited, only - # callbacks. Stays a plain thread-safe Future -> concurrent.futures.Future - # candidate, not create_future(). + # or on a user thread via manager.close()); never awaited, only callbacks. self._close_future = Future() self.in_flight_requests = collections.deque() self.broker_version_data = broker_version_data