diff options
| author | Dana Powers <dana.powers@gmail.com> | 2016-04-06 11:46:20 -0700 |
|---|---|---|
| committer | Dana Powers <dana.powers@gmail.com> | 2016-04-06 11:46:20 -0700 |
| commit | 358b4820744c42d47171f17a5b373a1c89f520bb (patch) | |
| tree | ca96eb3c0cab943b3c8931a24b1147f3f206aec3 | |
| parent | 452e7c2190b83f320f58e7f650302696dde458ed (diff) | |
| download | kafka-python-358b4820744c42d47171f17a5b373a1c89f520bb.tar.gz | |
Log debug messages when skipping fetched messages due to offset checks
| -rw-r--r-- | kafka/consumer/fetcher.py | 5 |
1 files changed, 5 insertions, 0 deletions
diff --git a/kafka/consumer/fetcher.py b/kafka/consumer/fetcher.py index c1f98eb..375090a 100644 --- a/kafka/consumer/fetcher.py +++ b/kafka/consumer/fetcher.py @@ -338,6 +338,8 @@ class Fetcher(six.Iterator): for record in self._unpack_message_set(tp, messages): # Fetched compressed messages may include additional records if record.offset < fetch_offset: + log.debug("Skipping message offset: %s (expecting %s)", + record.offset, fetch_offset) continue drained[tp].append(record) else: @@ -419,6 +421,9 @@ class Fetcher(six.Iterator): # Compressed messagesets may include earlier messages # It is also possible that the user called seek() elif msg.offset != self._subscriptions.assignment[tp].position: + log.debug("Skipping message offset: %s (expecting %s)", + msg.offset, + self._subscriptions.assignment[tp].position) continue self._subscriptions.assignment[tp].position = msg.offset + 1 |