Warn if coro did not complete before deadline

kafka/consumer/group.py

@@ -695,7 +695,9 @@ def commit(self, offsets=None, timeout_ms=None):
             IncompatibleBrokerVersion: if broker version < 0.8.1
             IllegalStateError: if group_id is None
             KafkaTimeoutError: if the commit does not complete within timeout_ms
-                (default default_api_timeout_ms).
+                (default default_api_timeout_ms). Note that a timeout does not
+                guarantee the commit did not happen -- it may still take effect
+                on the broker after this call returns.
         """
         if self.config['api_version'] < (0, 8, 1):
             raise Errors.IncompatibleBrokerVersion('Requires >= Kafka 0.8.1')

kafka/coordinator/consumer.py

@@ -832,7 +832,9 @@ def commit_offsets_sync(self, offsets, timeout_ms=None):
 
         Raises:
             KafkaTimeoutError: if the commit does not complete within timeout_ms
-                (default default_api_timeout_ms).
+                (default default_api_timeout_ms). Note that a timeout does not
+                guarantee the commit did not happen -- it may still take effect
+                on the broker after this call returns.
         """
         if not self._use_offset_apis:
             raise Errors.UnsupportedVersionError('OffsetCommitRequest requires 0.8.1+ broker')

kafka/net/selector.py

@@ -340,11 +340,17 @@ def _bridge_deadline_secs(self, timeout_ms):
     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(
+        exc = Errors.KafkaTimeoutError(
             'net.run(%s) did not complete within %d ms (+%d ms grace). The IO '
             'event loop may be stalled by blocking work on the IO thread '
             '(e.g. a synchronous rebalance listener/assignor).'
             % (name, op_ms, self.config['bridge_grace_ms']))
+        # A caught KafkaTimeoutError is otherwise invisible; surface the stall so
+        # operators can see it. The coroutine keeps running until it completes or
+        # hits its own deadline, so any side effects (e.g. an offset commit) may
+        # still take effect (see the late-completion warning in run()).
+        log.warning('%s', exc)
+        return exc
 
     def run(self, coro, *args, timeout_ms=None):
         """Schedules coro on the event loop, blocks until complete, returns value or raises.
@@ -359,7 +365,12 @@ def run(self, coro, *args, timeout_ms=None):
         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).
+        for a stalled IO loop; the coroutine itself is **not cancelled** -- it
+        keeps running until it completes or hits its own deadline (see #3121).
+        Consequently a timeout does not guarantee the operation did not happen:
+        a side-effecting coroutine (e.g. an offset commit) may still take effect
+        after the caller has seen the timeout. Such late completions are logged
+        at WARNING.
         """
         if self._closed:
             raise RuntimeError('NetworkSelector closed!')
@@ -382,6 +393,7 @@ def run(self, coro, *args, timeout_ms=None):
         elif self._exception:
             raise self._exception from None
 
+        coro_name = getattr(coro, '__name__', None) or repr(coro)
         event = threading.Event()
         state = {'value': None, 'exception': None}
         async def waiter():
@@ -396,14 +408,29 @@ async def waiter():
             finally:
                 with self._pending_waiters_lock:
                     self._pending_waiters.pop(event, None)
+                    abandoned = state.get('abandoned', False)
+                # The caller already gave up (backstop timeout) but the coroutine
+                # ran to completion anyway -- make the late outcome visible, since
+                # a late success means its side effects took effect after the
+                # caller saw a timeout.
+                if abandoned:
+                    if state['exception'] is None:
+                        log.warning('net.run(%s) completed successfully after the caller '
+                                    'timed out; its side effects took effect late.', coro_name)
+                    else:
+                        log.warning('net.run(%s) failed after the caller timed out: %s',
+                                    coro_name, state['exception'])
                 event.set()
         with self._pending_waiters_lock:
             self._pending_waiters[event] = state
         self.call_soon_threadsafe(waiter)
         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.
+            # `finally` pops _pending_waiters and (seeing 'abandoned') logs the
+            # late outcome before setting the (now-ignored) event.
+            with self._pending_waiters_lock:
+                state['abandoned'] = True
             raise self._bridge_timeout(coro, timeout_ms)
         if state['exception'] is not None:
             raise state['exception']  # pylint: disable=E0702

test/net/test_selector.py

@@ -1060,3 +1060,58 @@ async def waits_forever():
             assert 0.1 <= elapsed < 2.0
         finally:
             net.close()
+
+    def test_backstop_fire_logs_warning(self, caplog):
+        """A caught KafkaTimeoutError is otherwise invisible; the backstop logs
+        a WARNING naming the coroutine so a stall is observable."""
+        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)
+            with caplog.at_level('WARNING', logger='kafka.net.selector'):
+                th, outcome = self._run_in_thread(net, work)
+            assert isinstance(outcome.get('exc'), KafkaTimeoutError)
+            assert any('did not complete within' in r.message and 'work' in r.message
+                       for r in caplog.records), (
+                'expected a backstop WARNING, got: %r'
+                % [r.message for r in caplog.records])
+        finally:
+            release.set()
+            net.close()
+
+    def test_late_completion_of_abandoned_coroutine_logs(self, caplog):
+        """After the caller times out, if the coroutine still runs to completion
+        its late outcome is logged -- a late success means its side effects took
+        effect after the caller saw a timeout."""
+        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)
+            with caplog.at_level('WARNING', logger='kafka.net.selector'):
+                th, outcome = self._run_in_thread(net, work)  # caller times out
+                assert isinstance(outcome.get('exc'), KafkaTimeoutError)
+                # Release the wedge so the abandoned coroutine now completes.
+                release.set()
+                deadline = time.monotonic() + 2.0
+                while time.monotonic() < deadline:
+                    if any('after the caller timed out' in r.message
+                           for r in caplog.records):
+                        break
+                    time.sleep(0.01)
+            assert any('completed successfully after the caller timed out' in r.message
+                       and 'work' in r.message for r in caplog.records), (
+                'expected a late-completion WARNING, got: %r'
+                % [r.message for r in caplog.records])
+        finally:
+            release.set()
+            net.close()