Add default_api_timeout_ms config; raise KafkaTimeoutError from user-side of net.run()

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'],

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

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']

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.
         """
 

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)

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']

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)

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)

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()

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 == {}