diff options
author | Michael P. Nitowski <mpnitowski@gmail.com> | 2018-04-18 20:31:14 -0400 |
---|---|---|
committer | Dana Powers <dana.powers@gmail.com> | 2018-04-18 17:31:14 -0700 |
commit | afc6346d05054faf75e30a01b9e41a9916e703a9 (patch) | |
tree | 362300e22f682eb92f44abc34485b81f8ead5b27 | |
parent | b6ffbaaab2a26f8a12db195630dd5fa0a0fb2a35 (diff) | |
download | kafka-python-afc6346d05054faf75e30a01b9e41a9916e703a9.tar.gz |
Force lz4 to disable Kafka-unsupported block linking when encoding (#1476)
-rw-r--r-- | kafka/codec.py | 14 |
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: |