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

…1476)

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: