LZ4 support in kafka 0.8/0.9 does not accept a ContentSize header

Author: dpkp

kafka/codec.py

@@ -229,21 +229,29 @@ def lz4_encode_old_kafka(payload):
     assert xxhash is not None
     data = lz4_encode(payload)
     header_size = 7
-    if isinstance(data[4], int):
-        flg = data[4]
-    else:
-        flg = ord(data[4])
+    flg = data[4]
+    if not isinstance(flg, int):
+        flg = ord(flg)
+
     content_size_bit = ((flg >> 3) & 1)
     if content_size_bit:
-        header_size += 8
+        # Old kafka does not accept the content-size field
+        # so we need to discard it and reset the header flag
+        flg -= 8
+        data = bytearray(data)
+        data[4] = flg
+        data = bytes(data)
+        payload = data[header_size+8:]
+    else:
+        payload = data[header_size:]
 
     # This is the incorrect hc
     hc = xxhash.xxh32(data[0:header_size-1]).digest()[-2:-1]  # pylint: disable-msg=no-member
 
     return b''.join([
         data[0:header_size-1],
         hc,
-        data[header_size:]
+        payload
     ])