Skip to content

More / updated debug logging for coordinator / consumer #2630

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 1 commit into from
May 21, 2025
Merged

More / updated debug logging for coordinator / consumer #2630

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 5 additions & 2 deletions kafka/consumer/group.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -699,20 +699,21 @@ def _poll_once(self, timer, max_records, update_offsets=True):
dict: Map of topic to list of records (may be empty).
"""
if not self._coordinator.poll(timeout_ms=timer.timeout_ms):
log.debug('poll: timeout during coordinator.poll(); returning early')
return {}

has_all_fetch_positions = self._update_fetch_positions(timeout_ms=timer.timeout_ms)

# If data is available already, e.g. from a previous network client
# poll() call to commit, then just return it immediately
records, partial = self._fetcher.fetched_records(max_records, update_offsets=update_offsets)
log.debug('Fetched records: %s, %s', records, partial)
log.debug('poll: fetched records: %s, %s', records, partial)
# Before returning the fetched records, we can send off the
# next round of fetches and avoid block waiting for their
# responses to enable pipelining while the user is handling the
# fetched records.
if not partial:
log.debug("Sending fetches")
log.debug("poll: Sending fetches")
futures = self._fetcher.send_fetches()
if len(futures):
self._client.poll(timeout_ms=0)
Expand All @@ -724,12 +725,14 @@ def _poll_once(self, timer, max_records, update_offsets=True):
# since the offset lookup may be backing off after a failure
poll_timeout_ms = min(timer.timeout_ms, self._coordinator.time_to_next_poll() * 1000)
if not has_all_fetch_positions:
log.debug('poll: do not have all fetch positions...')
poll_timeout_ms = min(poll_timeout_ms, self.config['retry_backoff_ms'])

self._client.poll(timeout_ms=poll_timeout_ms)
# after the long poll, we should check whether the group needs to rebalance
# prior to returning data so that the group can stabilize faster
if self._coordinator.need_rejoin():
log.debug('poll: coordinator needs rejoin; returning early')
return {}

records, _ = self._fetcher.fetched_records(max_records, update_offsets=update_offsets)
Expand Down
5 changes: 5 additions & 0 deletions kafka/coordinator/consumer.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -274,6 +274,7 @@ def poll(self, timeout_ms=None):
try:
self._invoke_completed_offset_commit_callbacks()
if not self.ensure_coordinator_ready(timeout_ms=timer.timeout_ms):
log.debug('coordinator.poll: timeout in ensure_coordinator_ready; returning early')
return False

if self.config['api_version'] >= (0, 9) and self._subscription.partitions_auto_assigned():
Expand All @@ -293,9 +294,11 @@ def poll(self, timeout_ms=None):
metadata_update = self._client.cluster.request_update()
self._client.poll(future=metadata_update, timeout_ms=timer.timeout_ms)
if not metadata_update.is_done:
log.debug('coordinator.poll: timeout updating metadata; returning early')
return False

if not self.ensure_active_group(timeout_ms=timer.timeout_ms):
log.debug('coordinator.poll: timeout in ensure_active_group; returning early')
return False

self.poll_heartbeat()
Expand Down Expand Up @@ -722,6 +725,7 @@ def _send_offset_commit_request(self, offsets):
return future

def _handle_offset_commit_response(self, offsets, future, send_time, response):
log.debug("Received OffsetCommitResponse: %s", response)
# TODO look at adding request_latency_ms to response (like java kafka)
if self._consumer_sensors:
self._consumer_sensors.commit_latency.record((time.time() - send_time) * 1000)
Expand Down Expand Up @@ -848,6 +852,7 @@ def _send_offset_fetch_request(self, partitions):
return future

def _handle_offset_fetch_response(self, future, response):
log.debug("Received OffsetFetchResponse: %s", response)
if response.API_VERSION >= 2 and response.error_code != Errors.NoError.errno:
error_type = Errors.for_code(response.error_code)
log.debug("Offset fetch failed: %s", error_type.__name__)
Expand Down
Loading