Function _detect_xerial_stream

kafka/codec.py:165–197

Detects if the data given might have been encoded with the blocking mode of the xerial snappy library. This mode writes a magic header of the format: +--------+--------------+------------+---------+--------+ | Marker | Magic String | Null / Pad | Version | Co

(payload)

def _detect_xerial_stream(payload):
    """Detects if the data given might have been encoded with the blocking mode
        of the xerial snappy library.

        This mode writes a magic header of the format:
            +--------+--------------+------------+---------+--------+
            | Marker | Magic String | Null / Pad | Version | Compat |
            +--------+--------------+------------+---------+--------+
            |  byte  |   c-string   |    byte    |  int32  | int32  |
            +--------+--------------+------------+---------+--------+
            |  -126  |   'SNAPPY'   |     \0     |         |        |
            +--------+--------------+------------+---------+--------+

        The pad appears to be to ensure that SNAPPY is a valid cstring
        The version is the version of this format as written by xerial,
        in the wild this is currently 1 as such we only support v1.

        Compat is there to claim the minimum supported version that
        can read a xerial block stream, presently in the wild this is
        1.
    """

    if len(payload) > 16:
        magic = struct.unpack('!' + _XERIAL_V1_FORMAT[:8], bytes(payload)[:8])
        version, compat = struct.unpack('!' + _XERIAL_V1_FORMAT[8:], bytes(payload)[8:16])
        # Until there is more than one way to do xerial blocking, the version + compat
        # fields can be ignored. Also some producers (i.e., redpanda) are known to
        # incorrectly encode these as little-endian, and that causes us to fail decoding
        # when we otherwise would have succeeded.
        # See https://github.com/dpkp/kafka-python/issues/2414
        if magic == _XERIAL_V1_HEADER[:8]:
            return True
    return False


def snappy_decode(payload):