diff --git a/kafka/cluster.py b/kafka/cluster.py
index e41ac59eb..ad62c2f2f 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:
@@ -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)
@@ -419,18 +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:
-                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 e89873b47..14a0f9ba4 100644
--- a/kafka/consumer/fetcher.py
+++ b/kafka/consumer/fetcher.py
@@ -250,7 +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
 
-        wakeup = Future()
+        wakeup = Future()  # plain future, not backend / no await
         def _wake(_):
             if not wakeup.is_done:
                 wakeup.success(None)
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 ecaee6985..5c94be734 100644
--- a/kafka/future.py
+++ b/kafka/future.py
@@ -8,6 +8,20 @@
 
 
 class Future:
+    """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) 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
 
@@ -157,10 +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):
-        if not self.is_done:
-            yield self
-        if self.exception:
-            raise self.exception
-        return self.value
diff --git a/kafka/net/backend.py b/kafka/net/backend.py
new file mode 100644
index 000000000..58cba7f2c
--- /dev/null
+++ b/kafka/net/backend.py
@@ -0,0 +1,61 @@
+"""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.
+
+    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/connection.py b/kafka/net/connection.py
index 2794e1a22..45ca27f8d 100644
--- a/kafka/net/connection.py
+++ b/kafka/net/connection.py
@@ -53,7 +53,9 @@ 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._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.
         self._close_future = Future()
         self.in_flight_requests = collections.deque()
         self.broker_version_data = broker_version_data
@@ -116,7 +118,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 +133,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 +589,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..55501a2ea 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)
 
@@ -410,9 +411,11 @@ 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
-        wrapper = 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:
                 wrapper.success(value)
@@ -421,15 +424,26 @@ 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.
+
+        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..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))
@@ -445,6 +465,16 @@ def reschedule(self, when, task):
         self.call_at(when, task)
         return task
 
+    def create_future(self):
+        """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 ``SelectorFuture`` (a ``Future`` with ``__await__``).
+        """
+        return SelectorFuture()
+
     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/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_backend_future.py b/test/net/test_backend_future.py
new file mode 100644
index 000000000..b38e16e05
--- /dev/null
+++ b/test/net/test_backend_future.py
@@ -0,0 +1,183 @@
+"""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
+from kafka.net.backend import 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 SelectorFuture, 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 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_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/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 33d59f642..036b6fa6a 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'
@@ -508,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 4f5707ee5..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()
 
@@ -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).
 
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__')