From 1d4251a9efa4c5466ba5095f3ba199bf082a72b5 Mon Sep 17 00:00:00 2001 From: Dana Powers Date: Sun, 22 May 2016 17:14:03 -0700 Subject: Use standard LZ4 framing for v1 messages / kafka 0.10 (#695) * LZ4 framing fixed in 0.10 / message v1 -- retain broken lz4 code for compatibility * lz4f does not support easy incremental decompression - raise RuntimeError * Update lz4 codec tests --- kafka/protocol/message.py | 8 ++++++-- 1 file changed, 6 insertions(+), 2 deletions(-) (limited to 'kafka/protocol') diff --git a/kafka/protocol/message.py b/kafka/protocol/message.py index 473ca56..78840fc 100644 --- a/kafka/protocol/message.py +++ b/kafka/protocol/message.py @@ -2,7 +2,8 @@ import io import time from ..codec import (has_gzip, has_snappy, has_lz4, - gzip_decode, snappy_decode, lz4_decode) + gzip_decode, snappy_decode, + lz4_decode, lz4_decode_old_kafka) from . import pickle from .struct import Struct from .types import ( @@ -116,7 +117,10 @@ class Message(Struct): raw_bytes = snappy_decode(self.value) elif codec == self.CODEC_LZ4: assert has_lz4(), 'LZ4 decompression unsupported' - raw_bytes = lz4_decode(self.value) + if self.magic == 0: + raw_bytes = lz4_decode_old_kafka(self.value) + else: + raw_bytes = lz4_decode(self.value) else: raise Exception('This should be impossible') -- cgit v1.2.1