Force lz4 to disable Kafka-unsupported block linking when encoding (#1476)

1 files changed, 13 insertions, 1 deletions

diff --git a/kafka/codec.py b/kafka/codec.py
index de15e79..4d180dd 100644
--- a/kafka/codec.py
+++ b/kafka/codec.py
@@ -18,6 +18,18 @@ except ImportError:
 
 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: