diff --git a/kafka/codec.py b/kafka/codec.py index de15e7928..4d180ddd3 100644 --- a/kafka/codec.py +++ b/kafka/codec.py @@ -18,6 +18,18 @@ try: import lz4.frame as lz4 + + def _lz4_compress(payload, **kwargs): + # Kafka does not support LZ4 dependent blocks + try: + # For lz4>=0.12.0 + kwargs.pop('block_linked', None) + return lz4.compress(payload, block_linked=False, **kwargs) + except TypeError: + # For earlier versions of lz4 + kwargs.pop('block_mode', None) + return lz4.compress(payload, block_mode=1, **kwargs) + except ImportError: lz4 = None @@ -202,7 +214,7 @@ def snappy_decode(payload): if lz4: - lz4_encode = lz4.compress # pylint: disable-msg=no-member + lz4_encode = _lz4_compress # pylint: disable-msg=no-member elif lz4f: lz4_encode = lz4f.compressFrame # pylint: disable-msg=no-member elif lz4framed: