From 2dfcec51285b494adfd17e039bf4852330109fa8 Mon Sep 17 00:00:00 2001 From: mnito Date: Wed, 18 Apr 2018 12:34:06 -0400 Subject: [PATCH 1/2] Force lz4 to disable Kafka-unsupported block linking when encoding --- kafka/codec.py | 8 +++++++- 1 file changed, 7 insertions(+), 1 deletion(-) diff --git a/kafka/codec.py b/kafka/codec.py index de15e7928..fdfbae18d 100644 --- a/kafka/codec.py +++ b/kafka/codec.py @@ -18,6 +18,12 @@ try: import lz4.frame as lz4 + + def _lz4_compress(payload, **kwargs): + kwargs.pop('block_linked', None) + # Kafka does not support block linked mode + return lz4.compress(payload, block_linked=False, **kwargs) + except ImportError: lz4 = None @@ -202,7 +208,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: From 2e5de9cef0c50a6d7047955085b51bfcc506a838 Mon Sep 17 00:00:00 2001 From: mnito Date: Wed, 18 Apr 2018 18:35:16 -0400 Subject: [PATCH 2/2] Support independent blocks for earlier versions of python-lz4 --- kafka/codec.py | 12 +++++++++--- 1 file changed, 9 insertions(+), 3 deletions(-) diff --git a/kafka/codec.py b/kafka/codec.py index fdfbae18d..4d180ddd3 100644 --- a/kafka/codec.py +++ b/kafka/codec.py @@ -20,9 +20,15 @@ import lz4.frame as lz4 def _lz4_compress(payload, **kwargs): - kwargs.pop('block_linked', None) - # Kafka does not support block linked mode - return lz4.compress(payload, block_linked=False, **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