Consumer: sleep in poll() if timeout, no records, and no fetchable partitions (#3039)

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

Author: dpkp

kafka/consumer/fetcher.py

@@ -173,8 +173,12 @@ def fetch_records(self, max_records=None, update_offsets=True, timeout_ms=None):
                 Only applies when no records are immediately available.
 
         Returns:
-            dict[TopicPartition, list[ConsumerRecord]]: records grouped by
-            partition; may be empty if no records arrived in the budget.
+            tuple[dict[TopicPartition, list[ConsumerRecord]], bool]:
+                ``(records, idle)``. ``idle`` is True when there were no
+                buffered records, no in-flight fetches, and no pending
+                offset-reset task -- i.e. nothing this fetcher could wait
+                on. Callers in that state should sleep before retrying
+                instead of busy-looping.
         """
         # Drain whatever's already buffered from prior fetch responses.
         records, partial = self.fetched_records(
@@ -184,7 +188,7 @@ def fetch_records(self, max_records=None, update_offsets=True, timeout_ms=None):
             self.send_fetches()
 
         if records:
-            return records
+            return records, False
 
         # No records yet. Block until either an in-flight fetch
         # completes (records may have arrived) or a pending offset-reset
@@ -196,7 +200,7 @@ def fetch_records(self, max_records=None, update_offsets=True, timeout_ms=None):
         if self._reset_task is not None and not self._reset_task.is_done:
             waited_on.append(self._reset_task)
         if not waited_on:
-            return records  # nothing pending; nothing to wait for
+            return records, True  # nothing pending; caller should sleep
 
         wakeup = Future()
         def _wake(_):
@@ -212,7 +216,7 @@ def _wake(_):
 
         records, _ = self.fetched_records(
             max_records, update_offsets=update_offsets)
-        return records
+        return records, False
 
     def send_fetches(self):
         """Send FetchRequests for all assigned partitions that do not already have

kafka/consumer/group.py

@@ -813,8 +813,16 @@ def _poll_once(self, timer, max_records, update_offsets=True):
         if self.config['group_id'] is not None:
             poll_timeout_ms = min(poll_timeout_ms, self._coordinator.time_to_next_poll() * 1000)
 
-        return self._fetcher.fetch_records(
+        records, idle = self._fetcher.fetch_records(
             max_records, update_offsets=update_offsets, timeout_ms=poll_timeout_ms)
+        if not records and idle and poll_timeout_ms > 0:
+            # Nothing fetchable and nothing in flight (no assignment, all
+            # paused, or no resolvable positions). Sleep up to poll_timeout_ms
+            # to avoid busy-looping; poll_timeout_ms is already capped by the
+            # coordinator's next-action deadline.
+            poll_timeout_ms = min(poll_timeout_ms, self.config['retry_backoff_ms'])
+            time.sleep(poll_timeout_ms / 1000)
+        return records
 
     def position(self, partition, timeout_ms=None):
         """Get the offset of the next record that will be fetched

test/consumer/test_consumer.py

@@ -2,6 +2,7 @@
 
 from kafka import ConsumerGroupMetadata, KafkaConsumer, TopicPartition
 from kafka.errors import KafkaConfigurationError, IllegalStateError
+from kafka.util import Timer
 
 
 def test_session_timeout_different_from_max_poll_timeout_raises():
@@ -92,3 +93,69 @@ def test_group_metadata_with_group_id_delegates_to_coordinator():
     assert gm.generation_id == -1
     assert gm.member_id == ''
     consumer.close()
+
+
+def _stub_poll_path(consumer, mocker, fetch_return):
+    """Patch out the bits of _poll_once we don't care about so the test can
+    focus on the fetch -> sleep handoff."""
+    mocker.patch.object(consumer._coordinator, 'poll', return_value=True)
+    mocker.patch.object(consumer, '_refresh_committed_offsets')
+    mocker.patch.object(consumer._fetcher, 'reset_offsets_if_needed')
+    mocker.patch.object(consumer._fetcher, 'maybe_validate_positions')
+    mocker.patch.object(consumer._fetcher, 'validate_offsets_if_needed')
+    mocker.patch.object(consumer._fetcher, 'fetch_records',
+                        return_value=fetch_return)
+
+
+def test_poll_once_sleeps_when_fetcher_idle(mocker):
+    """If the fetcher reports it has no work pending, _poll_once sleeps up to
+    poll_timeout_ms before returning - otherwise consumer.poll() busy-loops
+    on no-fetchable-partition consumers."""
+    consumer = KafkaConsumer(api_version=(0, 10, 0))
+    try:
+        _stub_poll_path(consumer, mocker, fetch_return=({}, True))
+        sleep = mocker.patch('kafka.consumer.group.time.sleep')
+
+        records = consumer._poll_once(Timer(1000), max_records=100)
+
+        assert records == {}
+        sleep.assert_called_once()
+        slept_secs = sleep.call_args[0][0]
+        # poll_timeout_ms is uncapped for no-group consumers, so we sleep
+        # roughly the full Timer budget (1.0s).
+        assert 0 < slept_secs <= 1.0
+    finally:
+        consumer.close()
+
+
+def test_poll_once_does_not_sleep_when_records_returned(mocker):
+    """fetch_records returned records - no sleep, caller can return them."""
+    consumer = KafkaConsumer(api_version=(0, 10, 0))
+    try:
+        tp_records = {TopicPartition('foo', 0): [object()]}
+        _stub_poll_path(consumer, mocker, fetch_return=(tp_records, False))
+        sleep = mocker.patch('kafka.consumer.group.time.sleep')
+
+        records = consumer._poll_once(Timer(1000), max_records=100)
+
+        assert records is tp_records
+        sleep.assert_not_called()
+    finally:
+        consumer.close()
+
+
+def test_poll_once_does_not_sleep_when_fetcher_waited_but_empty(mocker):
+    """fetch_records returned ({}, False) - it waited on in-flight work and
+    got nothing. Don't sleep; the next loop iteration will check the
+    completed fetches without delay."""
+    consumer = KafkaConsumer(api_version=(0, 10, 0))
+    try:
+        _stub_poll_path(consumer, mocker, fetch_return=({}, False))
+        sleep = mocker.patch('kafka.consumer.group.time.sleep')
+
+        records = consumer._poll_once(Timer(1000), max_records=100)
+
+        assert records == {}
+        sleep.assert_not_called()
+    finally:
+        consumer.close()

test/consumer/test_fetcher.py

@@ -521,6 +521,59 @@ def test_fetched_records(fetcher, topic, mocker):
     assert partial is False
 
 
+def test_fetch_records_idle_when_nothing_pending(fetcher, mocker):
+    """No buffered records, no in-flight fetches, no reset task -> idle=True
+    so the caller (KafkaConsumer._poll_once) can sleep instead of
+    busy-looping."""
+    # Suppress fetch dispatch: simulate "nothing fetchable" (paused, no
+    # assignment, etc.) without going through the manager.
+    mocker.patch.object(fetcher, 'send_fetches', return_value=None)
+    assert not fetcher._fetch_futures
+    assert fetcher._reset_task is None
+
+    records, idle = fetcher.fetch_records(timeout_ms=0)
+    assert records == {}
+    assert idle is True
+
+
+def test_fetch_records_not_idle_when_records_buffered(fetcher, topic):
+    """Buffered records short-circuit before the no-work check; idle=False."""
+    fetcher.config['check_crcs'] = False
+    tp = TopicPartition(topic, 0)
+    fetcher._completed_fetches.append(
+        _build_completed_fetch(tp, [(None, b'foo', None)]))
+
+    records, idle = fetcher.fetch_records(timeout_ms=0)
+    assert tp in records
+    assert idle is False
+
+
+def test_fetch_records_not_idle_when_inflight_fetch_pending(fetcher, mocker):
+    """An in-flight fetch future means there's work to wait on; idle=False
+    even if the wait times out with no records."""
+    pending = Future()
+    fetcher._fetch_futures.append(pending)
+    mocker.patch.object(fetcher, 'send_fetches', return_value=None)
+    # Don't actually block on the wakeup - just return as if the wait
+    # timed out.
+    mocker.patch.object(fetcher._net, 'run', return_value=None)
+
+    records, idle = fetcher.fetch_records(timeout_ms=0)
+    assert records == {}
+    assert idle is False
+
+
+def test_fetch_records_not_idle_when_reset_task_pending(fetcher, mocker):
+    """An in-flight offset-reset task counts as pending work; idle=False."""
+    fetcher._reset_task = Future()
+    mocker.patch.object(fetcher, 'send_fetches', return_value=None)
+    mocker.patch.object(fetcher._net, 'run', return_value=None)
+
+    records, idle = fetcher.fetch_records(timeout_ms=0)
+    assert records == {}
+    assert idle is False
+
+
 @pytest.mark.parametrize(("fetch_offsets", "fetch_response", "num_partitions"), [
     (
         {TopicPartition('foo', 0): 0},