Function lz4_encode_old_kafka

kafka/codec.py:259–288

Encode payload for 0.8/0.9 brokers -- requires an incorrect header checksum.

(payload)

def lz4_encode_old_kafka(payload):
    """Encode payload for 0.8/0.9 brokers -- requires an incorrect header checksum."""
    if xxhash is None:
        raise RuntimeError('pip install xxhash for lz4 encoding with 0.8/0.9 brokers')
    data = lz4_encode(payload)
    header_size = 7
    flg = data[4]
    if not isinstance(flg, int):
        flg = ord(flg)

    content_size_bit = ((flg >> 3) & 1)
    if content_size_bit:
        # 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,
        payload
    ])


def lz4_decode_old_kafka(payload):