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

Author: dpkp

kafka/codec.py

@@ -233,17 +233,27 @@ def lz4_encode_old_kafka(payload):
         flg = data[4]
     else:
         flg = ord(data[4])
+
     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
+        if isinstance(data[4], int):
+            data[4] = flg
+        else:
+            data = data[0:4] + chr(flg) + data[5:]
+        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
     ])