diff --git a/kafka/admin/client.py b/kafka/admin/client.py
index 824d0b284..f51b7682c 100644
--- a/kafka/admin/client.py
+++ b/kafka/admin/client.py
@@ -63,6 +63,11 @@ class KafkaAdminClient(
             20% below and 20% above the computed value. Default: 30000.
         request_timeout_ms (int): Client request timeout in milliseconds.
             Default: 30000.
+        default_api_timeout_ms (int): Default timeout in milliseconds for
+            blocking client APIs that do not take an explicit timeout. Bounds
+            the whole operation and serves as a liveness backstop so a stalled
+            IO loop cannot hang the calling thread indefinitely. Should be >=
+            request_timeout_ms. Default: 60000.
         connections_max_idle_ms: Close idle connections after the number of
             milliseconds specified by this config. The broker closes idle
             connections after connections.max.idle.ms, so this avoids hitting
@@ -183,6 +188,7 @@ class KafkaAdminClient(
         'bootstrap_servers': 'localhost',
         'client_id': 'kafka-python-' + __version__,
         'request_timeout_ms': 30000,
+        'default_api_timeout_ms': 60000,
         'connections_max_idle_ms': 9 * 60 * 1000,
         'reconnect_backoff_ms': 50,
         'reconnect_backoff_max_ms': 30000,
@@ -233,6 +239,11 @@ def __init__(self, **configs):
         self.config.pop('selector')
         self.config.update(configs)
 
+        if self.config['default_api_timeout_ms'] < self.config['request_timeout_ms']:
+            raise KafkaConfigurationError(
+                "default_api_timeout_ms ({}) must be >= request_timeout_ms ({})."
+                .format(self.config['default_api_timeout_ms'], self.config['request_timeout_ms']))
+
         # Configure metrics
         metrics_tags = {'client-id': self.config['client_id']}
         metric_config = MetricConfig(samples=self.config['metrics_num_samples'],
diff --git a/kafka/consumer/group.py b/kafka/consumer/group.py
index ed4530797..becd80ac2 100644
--- a/kafka/consumer/group.py
+++ b/kafka/consumer/group.py
@@ -103,6 +103,11 @@ class KafkaConsumer:
             message on a certain partition. Default: 1048576.
         request_timeout_ms (int): Client request timeout in milliseconds.
             Default: 305000.
+        default_api_timeout_ms (int): Default timeout in milliseconds for
+            blocking client APIs that do not take an explicit timeout (e.g.
+            commit()). Bounds the whole operation and serves as a liveness
+            backstop so a stalled IO loop cannot hang the calling thread
+            indefinitely. Should be >= request_timeout_ms. Default: 60000.
         retry_backoff_ms (int): Milliseconds to backoff when retrying on
             errors. Default: 100.
         reconnect_backoff_ms (int): The amount of time in milliseconds to
@@ -320,6 +325,7 @@ class KafkaConsumer:
         'fetch_max_bytes': 52428800,
         'max_partition_fetch_bytes': 1 * 1024 * 1024,
         'request_timeout_ms': 30000,
+        'default_api_timeout_ms': 60000,
         'retry_backoff_ms': 100,
         'reconnect_backoff_ms': 50,
         'reconnect_backoff_max_ms': 30000,
@@ -403,6 +409,12 @@ def __init__(self, *topics, **configs):
                 "fetch_max_wait_ms ({})."
                 .format(connections_max_idle_ms, request_timeout_ms, fetch_max_wait_ms))
 
+        default_api_timeout_ms = self.config['default_api_timeout_ms']
+        if default_api_timeout_ms < request_timeout_ms:
+            raise KafkaConfigurationError(
+                "default_api_timeout_ms ({}) must be >= request_timeout_ms ({})."
+                .format(default_api_timeout_ms, request_timeout_ms))
+
         # fetch_max_bytes (KIP-74) is a soft cap the broker applies to the
         # *record data* in a FetchResponse body, not to the whole frame. The
         # frame also carries the response header + per-topic/per-partition
diff --git a/kafka/net/asyncio_backend.py b/kafka/net/asyncio_backend.py
index 4cbbf9cdf..ef8cc4c20 100644
--- a/kafka/net/asyncio_backend.py
+++ b/kafka/net/asyncio_backend.py
@@ -94,6 +94,12 @@ def __init__(self, *, loop=None, loop_factory=None, **configs):
         self._io_thread = None
         self._closed = False
         self._client_id = configs.get('client_id') or 'kafka-python'
+        # See NetworkSelector: default operation deadline + grace margin for the
+        # cross-thread run() liveness backstop (#3121).
+        self._default_api_timeout_ms = configs.get('default_api_timeout_ms') or 60000
+        self._bridge_grace_ms = configs.get('bridge_grace_ms')
+        if self._bridge_grace_ms is None:
+            self._bridge_grace_ms = 5000
         # Strong refs to live tasks (asyncio only holds weak refs, so bare
         # tasks can be GC'd mid-flight); mirrors NetworkSelector._pending_tasks.
         self._pending = set()
@@ -287,7 +293,7 @@ async def wrapper():
         return future
 
     # --- cross-thread bridge ---------------------------------------------
-    def run(self, coro, *args):
+    def run(self, coro, *args, timeout_ms=None):
         if self._closed:
             raise RuntimeError('AsyncioBackend closed!')
         if self._io_thread is None:
@@ -299,6 +305,8 @@ def run(self, coro, *args):
                 "(or another IO-thread callback) calls a blocking consumer/admin API. "
                 "Use AsyncConsumerRebalanceListener and await the async variant, "
                 "or move the blocking work to a worker thread.")
+        op_ms = timeout_ms if timeout_ms is not None else self._default_api_timeout_ms
+        deadline_secs = (op_ms + self._bridge_grace_ms) / 1000
         event = threading.Event()
         state = {'value': None, 'exception': None}
 
@@ -316,7 +324,15 @@ async def waiter():
         with self._pending_waiters_lock:
             self._pending_waiters[event] = state
         self.call_soon(waiter)
-        event.wait()
+        if not event.wait(timeout=deadline_secs):
+            # Loop never ran the coroutine to completion; leave the waiter
+            # registered (its finally pops it) and surface a liveness timeout.
+            name = getattr(coro, '__name__', None) or repr(coro)
+            raise Errors.KafkaTimeoutError(
+                'net.run(%s) did not complete within %d ms (+%d ms grace). The '
+                'IO event loop may be stalled by blocking work on the IO thread '
+                '(e.g. a synchronous rebalance listener/assignor).'
+                % (name, op_ms, self._bridge_grace_ms))
         if state['exception'] is not None:
             raise state['exception']  # pylint: disable=raising-bad-type
         return state['value']
diff --git a/kafka/net/backend.py b/kafka/net/backend.py
index 4b7ad7bfb..823b0149a 100644
--- a/kafka/net/backend.py
+++ b/kafka/net/backend.py
@@ -199,9 +199,14 @@ async def create_connection(
         """
 
     # --- cross-thread bridge ---------------------------------------------
-    def run(self, coro: Any, *args: Any) -> Any:
+    def run(self, coro: Any, *args: Any, timeout_ms: Optional[float] = None) -> Any:
         """Schedule ``coro`` on the loop, block the calling thread, return/raise.
 
+        The blocking wait is bounded: if the coroutine does not complete within
+        ``timeout_ms`` (or the backend's ``default_api_timeout_ms`` when None),
+        plus a grace margin, ``KafkaTimeoutError`` is raised as a liveness
+        backstop against a stalled IO loop.
+
         Raises ``RuntimeError`` if called from the IO thread itself.
         """
 
diff --git a/kafka/net/manager.py b/kafka/net/manager.py
index 68091ab26..11ff89412 100644
--- a/kafka/net/manager.py
+++ b/kafka/net/manager.py
@@ -28,6 +28,7 @@ class KafkaConnectionManager:
         'reconnect_backoff_ms': 50,
         'reconnect_backoff_max_ms': 30000,
         'request_timeout_ms': 30000,
+        'default_api_timeout_ms': 60000,
         'socket_connection_setup_timeout_ms': 10000,
         'socket_connection_setup_timeout_max_ms': 30000,
         'socket_options': [
@@ -72,6 +73,11 @@ def __init__(self, net=None, **configs):
                 "client_dns_lookup must be one of %s; got %r"
                 % (self._VALID_DNS_LOOKUP_MODES, self.config['client_dns_lookup']))
 
+        if self.config['default_api_timeout_ms'] < self.config['request_timeout_ms']:
+            raise Errors.KafkaConfigurationError(
+                "default_api_timeout_ms (%s) must be >= request_timeout_ms (%s)"
+                % (self.config['default_api_timeout_ms'], self.config['request_timeout_ms']))
+
         if configs.get('socks5_proxy') is not None:
             if self.config['proxy_url'] is None:
                 log.warning('socks5_proxy is deprecated, use proxy_url instead')
@@ -467,7 +473,7 @@ def call_soon(self, coro, *args):
         """
         return self._net.call_soon_with_future(coro, *args)
 
-    def run(self, coro, *args):
+    def run(self, coro, *args, timeout_ms=None):
         """Schedules coro on the event loop, blocks until complete, returns value or raises.
 
         If an IO thread is running (via start()), the caller thread blocks on
@@ -476,6 +482,10 @@ def run(self, coro, *args):
 
         If no IO thread is running, falls back to driving the loop on the
         caller thread (legacy behavior).
+
+        The blocking wait is bounded by ``timeout_ms`` (or the backend's
+        ``default_api_timeout_ms`` when None) plus a grace margin; see
+        :meth:`NetworkSelector.run`.
         """
         self._maybe_start()
-        return self._net.run(coro, *args)
+        return self._net.run(coro, *args, timeout_ms=timeout_ms)
diff --git a/kafka/net/selector.py b/kafka/net/selector.py
index 3ec6dc30e..6f96f7916 100644
--- a/kafka/net/selector.py
+++ b/kafka/net/selector.py
@@ -207,6 +207,18 @@ class NetworkSelector:
         # When True, raise RuntimeError on slow tasks instead of just warning.
         # Useful in tests so livelocks fail loudly.
         'raise_on_slow_task': False,
+        # Default operation deadline for a cross-thread run() call that does not
+        # pass its own timeout_ms. Bounds the caller's blocking wait so a stalled
+        # IO loop (e.g. a synchronous rebalance listener/assignor or a blocking
+        # DNS lookup on the IO thread) cannot hang the caller indefinitely.
+        # Mirrors the Java client's default.api.timeout.ms.
+        'default_api_timeout_ms': 60000,
+        # Extra slack added on top of the operation deadline before run()'s
+        # cross-thread wait gives up. The coroutine enforces the operation
+        # deadline itself on a healthy loop; this margin ensures that self-timeout
+        # (and its unwind/retry-backoff) resolves and wins the race, so the
+        # backstop only trips when the loop genuinely isn't running the coroutine.
+        'bridge_grace_ms': 5000,
     }
 
     def __init__(self, **configs):
@@ -313,7 +325,26 @@ def _fail_pending_waiters(self, exc):
             state['exception'] = exc
             event.set()
 
-    def run(self, coro, *args):
+    def _bridge_deadline_secs(self, timeout_ms):
+        """Wall-clock ceiling for a cross-thread run() wait, in seconds.
+
+        The operation deadline defaults to ``default_api_timeout_ms`` when the
+        caller passes ``timeout_ms=None``; ``bridge_grace_ms`` is added so the
+        coroutine's own (equal) deadline wins the race on a healthy loop.
+        """
+        op_ms = timeout_ms if timeout_ms is not None else self.config['default_api_timeout_ms']
+        return (op_ms + self.config['bridge_grace_ms']) / 1000
+
+    def _bridge_timeout(self, coro, timeout_ms):
+        op_ms = timeout_ms if timeout_ms is not None else self.config['default_api_timeout_ms']
+        name = getattr(coro, '__name__', None) or repr(coro)
+        return Errors.KafkaTimeoutError(
+            'net.run(%s) did not complete within %d ms (+%d ms grace). The IO '
+            'event loop may be stalled by blocking work on the IO thread '
+            '(e.g. a synchronous rebalance listener/assignor).'
+            % (name, op_ms, self.config['bridge_grace_ms']))
+
+    def run(self, coro, *args, timeout_ms=None):
         """Schedules coro on the event loop, blocks until complete, returns value or raises.
 
         If an IO thread is running (via start()), the caller thread blocks on
@@ -322,12 +353,20 @@ def run(self, coro, *args):
 
         If no IO thread is running, falls back to driving the loop on the
         caller thread (legacy behavior).
+
+        The blocking wait is always bounded: if the coroutine does not complete
+        within ``timeout_ms`` (or ``default_api_timeout_ms`` when None), plus a
+        grace margin, KafkaTimeoutError is raised. This is a liveness backstop
+        for a stalled IO loop; the coroutine itself is left running (see #3121).
         """
         if self._closed:
             raise RuntimeError('NetworkSelector closed!')
+        deadline_secs = self._bridge_deadline_secs(timeout_ms)
         if self._io_thread is None:
             future = self.call_soon_with_future(coro, *args)
-            self.poll(future=future)
+            self.poll(timeout_ms=deadline_secs * 1000, future=future)
+            if not future.is_done:
+                raise self._bridge_timeout(coro, timeout_ms)
             if future.exception is not None:
                 raise future.exception
             return future.value
@@ -359,7 +398,11 @@ async def waiter():
         with self._pending_waiters_lock:
             self._pending_waiters[event] = state
         self.call_soon_threadsafe(waiter)
-        event.wait()
+        if not event.wait(timeout=deadline_secs):
+            # Loop never ran the coroutine to completion within the deadline.
+            # Leave the waiter registered: if the coroutine later finishes, its
+            # `finally` pops _pending_waiters and sets the (now-ignored) event.
+            raise self._bridge_timeout(coro, timeout_ms)
         if state['exception'] is not None:
             raise state['exception']  # pylint: disable=E0702
         return state['value']
diff --git a/kafka/producer/kafka.py b/kafka/producer/kafka.py
index 1eafbab61..1a3bee1a9 100644
--- a/kafka/producer/kafka.py
+++ b/kafka/producer/kafka.py
@@ -280,6 +280,11 @@ class KafkaProducer:
             errors. Default: 100.
         request_timeout_ms (int): Client request timeout in milliseconds.
             Default: 30000.
+        default_api_timeout_ms (int): Default timeout in milliseconds for
+            blocking client APIs that do not take an explicit timeout. Bounds
+            the whole operation and serves as a liveness backstop so a stalled
+            IO loop cannot hang the calling thread indefinitely. Should be >=
+            request_timeout_ms. Default: 60000.
         receive_message_max_bytes (int): Maximum allowed network frame size.
             Used to avoid OOM when decoding malformed network message header.
             Default: 100_000_000.
@@ -435,6 +440,7 @@ class KafkaProducer:
         'client_dns_lookup': 'use_all_dns_ips',
         'retry_backoff_ms': 100,
         'request_timeout_ms': 30000,
+        'default_api_timeout_ms': 60000,
         'receive_message_max_bytes': 100_000_000,
         'receive_buffer_bytes': None,
         'send_buffer_bytes': None,
@@ -489,6 +495,11 @@ def __init__(self, **configs):
         self.config.pop('selector')
         self.config.update(configs)
 
+        if self.config['default_api_timeout_ms'] < self.config['request_timeout_ms']:
+            raise Errors.KafkaConfigurationError(
+                "default_api_timeout_ms ({}) must be >= request_timeout_ms ({})."
+                .format(self.config['default_api_timeout_ms'], self.config['request_timeout_ms']))
+
         for key in ('key_serializer', 'value_serializer'):
             if self.config[key] is not None and not isinstance(self.config[key], Serializer):
                 warnings.warn('%s does not implement kafka.serializer.Serializer' % (key,), category=DeprecationWarning, stacklevel=3)
diff --git a/test/admin/test_admin_client_config.py b/test/admin/test_admin_client_config.py
new file mode 100644
index 000000000..ee750d952
--- /dev/null
+++ b/test/admin/test_admin_client_config.py
@@ -0,0 +1,11 @@
+import pytest
+
+from kafka.admin import KafkaAdminClient
+from kafka.errors import KafkaConfigurationError
+
+
+def test_default_api_timeout_smaller_than_request_timeout_raises():
+    # Validation runs before bootstrap/network, so no broker is needed.
+    with pytest.raises(KafkaConfigurationError):
+        KafkaAdminClient(bootstrap_servers='localhost:9092',
+                         request_timeout_ms=70000, default_api_timeout_ms=60000)
diff --git a/test/consumer/test_consumer.py b/test/consumer/test_consumer.py
index 5d99d9efe..2026978c4 100644
--- a/test/consumer/test_consumer.py
+++ b/test/consumer/test_consumer.py
@@ -20,6 +20,12 @@ def test_request_timeout_larger_than_connections_max_idle_ms_raises():
         KafkaConsumer(bootstrap_servers='localhost:9092', api_version=(0, 9), request_timeout_ms=50000, connections_max_idle_ms=40000)
 
 
+def test_default_api_timeout_smaller_than_request_timeout_raises():
+    with pytest.raises(KafkaConfigurationError):
+        KafkaConsumer(bootstrap_servers='localhost:9092', api_version=(0, 9),
+                      request_timeout_ms=70000, default_api_timeout_ms=60000)
+
+
 def test_subscription_copy():
     consumer = KafkaConsumer('foo', api_version=(0, 10, 0))
     sub = consumer.subscription()
diff --git a/test/net/test_manager.py b/test/net/test_manager.py
index 99450391c..adaf71f2c 100644
--- a/test/net/test_manager.py
+++ b/test/net/test_manager.py
@@ -46,6 +46,25 @@ def test_config_override(self, net):
         m = KafkaConnectionManager(net, reconnect_backoff_ms=100)
         assert m.config['reconnect_backoff_ms'] == 100
 
+    def test_default_api_timeout_ms_default(self, net):
+        m = KafkaConnectionManager(net)
+        assert m.config['default_api_timeout_ms'] == 60000
+
+    def test_default_api_timeout_ms_flows_to_net_backend(self):
+        # net=None -> the manager constructs its own backend from config, so the
+        # user-supplied default_api_timeout_ms must reach it (issue #3121).
+        # Must satisfy default_api_timeout_ms >= request_timeout_ms (default 30000).
+        m = KafkaConnectionManager(default_api_timeout_ms=90000)
+        try:
+            assert m.config['default_api_timeout_ms'] == 90000
+            assert m._net.config['default_api_timeout_ms'] == 90000
+        finally:
+            m.close()
+
+    def test_default_api_timeout_ms_smaller_than_request_timeout_raises(self, net):
+        with pytest.raises(Errors.KafkaConfigurationError, match='default_api_timeout_ms'):
+            KafkaConnectionManager(net, request_timeout_ms=70000, default_api_timeout_ms=60000)
+
     def test_initial_state(self, net):
         manager = KafkaConnectionManager(net)
         assert manager._conns == {}
diff --git a/test/net/test_selector.py b/test/net/test_selector.py
index e4fbbf886..debc9c96e 100644
--- a/test/net/test_selector.py
+++ b/test/net/test_selector.py
@@ -932,3 +932,131 @@ def _poll_once_raising(*args, **kwargs):
         # Restore and verify the lock was released so the next poll succeeds.
         net._poll_once = orig
         net.poll(timeout_ms=10)  # would raise 'Concurrent access' if leaked
+
+
+class TestRunBridgeBackstop:
+    """Issue #3121: NetworkSelector.run() must bound its cross-thread wait so a
+    stalled IO loop can't hang a caller thread forever.
+
+    The original bug: run() blocked on a bare event.wait() with no timeout, so a
+    user-thread commit()/poll() hung indefinitely whenever the IO loop stopped
+    making progress (a blocking sync rebalance listener/assignor, or a blocking
+    DNS lookup on the IO thread). run() now bounds the wait by timeout_ms (or
+    default_api_timeout_ms when None) plus a grace margin, and raises
+    KafkaTimeoutError as a liveness backstop.
+    """
+
+    @staticmethod
+    def _wedge(net, release):
+        """Occupy the single IO thread until released; returns (wedged, release)."""
+        wedged = threading.Event()
+
+        async def wedge():
+            wedged.set()
+            release.wait(timeout=5.0)  # safety cap so the suite can't hang
+
+        net.call_soon_threadsafe(wedge)
+        assert wedged.wait(timeout=1.0), 'IO thread never entered the wedge'
+
+    def _run_in_thread(self, net, coro, **kw):
+        outcome = {}
+
+        def caller():
+            start = time.monotonic()
+            try:
+                outcome['value'] = net.run(coro, **kw)
+            except BaseException as exc:  # noqa: B036 - record whatever surfaces
+                outcome['exc'] = exc
+            outcome['elapsed'] = time.monotonic() - start
+
+        th = threading.Thread(target=caller, daemon=True)
+        th.start()
+        th.join(timeout=2.0)
+        return th, outcome
+
+    def test_default_deadline_bounds_wedged_loop(self):
+        """With no explicit timeout_ms, run() is still bounded by
+        default_api_timeout_ms (+grace) -- the core #3121 fix."""
+        net = NetworkSelector(default_api_timeout_ms=100, bridge_grace_ms=100)
+        net.start()
+
+        async def work():
+            return 'ok'
+
+        release = threading.Event()
+        try:
+            self._wedge(net, release)
+            th, outcome = self._run_in_thread(net, work)  # no timeout_ms
+            assert not th.is_alive(), 'run() hung past the default deadline (#3121)'
+            assert isinstance(outcome.get('exc'), KafkaTimeoutError), (
+                'expected KafkaTimeoutError, got: %r' % outcome)
+            assert 0.15 <= outcome['elapsed'] < 2.0
+            assert 'may be stalled' in str(outcome['exc'])
+        finally:
+            release.set()
+            net.close()
+
+    def test_explicit_timeout_bounds_wedged_loop(self):
+        """An explicit timeout_ms is honored even when the loop is wedged."""
+        net = NetworkSelector(default_api_timeout_ms=60000, bridge_grace_ms=100)
+        net.start()
+
+        async def work():
+            return 'ok'
+
+        release = threading.Event()
+        try:
+            self._wedge(net, release)
+            th, outcome = self._run_in_thread(net, work, timeout_ms=100)
+            assert not th.is_alive()
+            assert isinstance(outcome.get('exc'), KafkaTimeoutError)
+            assert 0.15 <= outcome['elapsed'] < 2.0
+        finally:
+            release.set()
+            net.close()
+
+    def test_healthy_loop_returns_value(self):
+        """The backstop must not perturb the normal path."""
+        net = NetworkSelector(default_api_timeout_ms=100, bridge_grace_ms=100)
+        net.start()
+
+        async def work():
+            return 42
+
+        try:
+            assert net.run(work) == 42
+        finally:
+            net.close()
+
+    def test_healthy_loop_propagates_coroutine_exception(self):
+        """A coroutine that raises surfaces its own exception, not the backstop's
+        generic timeout -- the backstop only fires on non-completion."""
+        net = NetworkSelector(default_api_timeout_ms=100, bridge_grace_ms=100)
+        net.start()
+
+        async def boom():
+            raise ValueError('from coroutine')
+
+        try:
+            with pytest.raises(ValueError, match='from coroutine'):
+                net.run(boom)
+        finally:
+            net.close()
+
+    def test_no_io_thread_fallback_honors_deadline(self):
+        """The no-IO-thread fallback path (drives the loop inline) also bounds
+        the wait -- previously it called poll(future=...) with no timeout."""
+        net = NetworkSelector(default_api_timeout_ms=100, bridge_grace_ms=50)
+        never = net.create_future()  # never resolved
+
+        async def waits_forever():
+            return await never
+
+        try:
+            start = time.monotonic()
+            with pytest.raises(KafkaTimeoutError):
+                net.run(waits_forever)  # no start() -> fallback path
+            elapsed = time.monotonic() - start
+            assert 0.1 <= elapsed < 2.0
+        finally:
+            net.close()
diff --git a/test/producer/test_producer.py b/test/producer/test_producer.py
index 25d44c68d..9ee013c7c 100644
--- a/test/producer/test_producer.py
+++ b/test/producer/test_producer.py
@@ -6,12 +6,20 @@
 import pytest
 
 from kafka import KafkaProducer
+from kafka.errors import KafkaConfigurationError
 from kafka.partitioner import Partitioner, StickyPartitioner
 from kafka.producer.transaction_manager import TransactionManager, ProducerIdAndEpoch
 
 from test.mock_broker import MockBroker
 
 
+def test_default_api_timeout_smaller_than_request_timeout_raises():
+    # Validation runs before any network/bootstrap, so no broker is needed.
+    with pytest.raises(KafkaConfigurationError):
+        KafkaProducer(bootstrap_servers='localhost:9092', api_version=(0, 9),
+                      request_timeout_ms=70000, default_api_timeout_ms=60000)
+
+
 def _mock_producer(**configs):
     """A KafkaProducer wired to a fresh MockBroker (no real network).