diff --git a/kafka/consumer/group.py b/kafka/consumer/group.py index ac52f7f06..3804147ab 100644 --- a/kafka/consumer/group.py +++ b/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') diff --git a/kafka/coordinator/consumer.py b/kafka/coordinator/consumer.py index dfa26057f..666d8aaf2 100644 --- a/kafka/coordinator/consumer.py +++ b/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') diff --git a/kafka/net/selector.py b/kafka/net/selector.py index 8e9a621ca..bd473abb6 100644 --- a/kafka/net/selector.py +++ b/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,6 +408,18 @@ 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 @@ -403,7 +427,10 @@ async def 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 diff --git a/test/net/test_selector.py b/test/net/test_selector.py index debc9c96e..dd52e62e5 100644 --- a/test/net/test_selector.py +++ b/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()