diff options
| -rw-r--r-- | kafka/consumer/fetcher.py | 8 |
1 files changed, 8 insertions, 0 deletions
diff --git a/kafka/consumer/fetcher.py b/kafka/consumer/fetcher.py index 7437567..62e28d6 100644 --- a/kafka/consumer/fetcher.py +++ b/kafka/consumer/fetcher.py @@ -361,6 +361,14 @@ class Fetcher(six.Iterator): # If relative offset is used, we need to decompress the entire message first to compute # the absolute offset. inner_mset = msg.decompress() + + # There should only ever be a single layer of compression + if inner_mset[0][-1].is_compressed(): + log.warning('MessageSet at %s offset %d appears ' + ' double-compressed. This should not' + ' happen -- check your producers!', + tp, offset) + if msg.magic > 0: last_offset, _, _ = inner_mset[-1] absolute_base_offset = offset - last_offset |