From 4cbc1e33747de22ac1e6ea673aea2f2d88b1b632 Mon Sep 17 00:00:00 2001
From: Dana Powers <dana.powers@gmail.com>
Date: Sun, 22 May 2016 15:02:07 -0700
Subject: [PATCH] lz4f does not support easy incremental decompression - raise
 RuntimeError

---
 kafka/codec.py | 11 +++++++++--
 1 file changed, 9 insertions(+), 2 deletions(-)

diff --git a/kafka/codec.py b/kafka/codec.py
index 8853408..9c31e9d 100644
--- a/kafka/codec.py
+++ b/kafka/codec.py
@@ -181,13 +181,20 @@ def snappy_decode(payload):
 
 def lz4_encode(payload):
     """Encode payload using interoperable LZ4 framing. Requires Kafka >= 0.10"""
+    # pylint: disable-msg=no-member
     return lz4f.compressFrame(payload)
 
 
 def lz4_decode(payload):
     """Decode payload using interoperable LZ4 framing. Requires Kafka >= 0.10"""
-    cCtx = lz4f.createCompContext()  # pylint: disable-msg=no-member
-    data = lz4f.decompressFrame(payload, cCtx)  # pylint: disable-msg=no-member
+    # pylint: disable-msg=no-member
+    ctx = lz4f.createDecompContext()
+    data = lz4f.decompressFrame(payload, ctx)
+
+    # lz4f python module does not expose how much of the payload was
+    # actually read if the decompression was only partial.
+    if data['next'] != 0:
+        raise RuntimeError('lz4f unable to decompress full payload')
     return data['decomp']