anchor.diff:  %% Return messages in reversed order.
   %% @end
   -spec decode_message_stream(binary(), Decoded) -> Decoded
  -        when Decoded :: [kpro_Message() | incomplete_message()].
  +        when Decoded :: [kpro_message() | incomplete_message()].
   decode_message_stream(<<>>, Acc) ->
     %% Do not reverse here!
     %% as the input is recursive when compressed
     Acc;
  +% FIXME cleanup
   decode_message_stream(Bin, Acc) ->
     {Msg, Rest} = decode_message(Bin),
     NewAcc =
       case Msg of
  -      #kpro_Message{attributes = Attr} = Msg when ?KPRO_IS_GZIP_ATTR(Attr) ->
  -        decode_message_stream(zlib:gunzip(Msg#kpro_Message.value), Acc);
  -      #kpro_Message{attributes = Attr} = Msg when ?KPRO_IS_SNAPPY_ATTR(Attr) ->
  -        decode_message_stream(java_snappy_unpack(Msg#kpro_Message.value), Acc);
  -      #kpro_Message{attributes = Attr} = Msg when ?KPRO_IS_LZ4_ATTR(Attr) ->
  -        decode_message_stream(lz4_unpack(Msg#kpro_Message.value), Acc);
  +      #kpro_message{attributes = Attr} = Msg when ?KPRO_IS_GZIP_ATTR(Attr) ->
  +        decode_message_stream(zlib:gunzip(Msg#kpro_message.value), Acc);
  +      #kpro_message{attributes = Attr} = Msg when ?KPRO_IS_SNAPPY_ATTR(Attr) ->
  +        decode_message_stream(java_snappy_unpack(Msg#kpro_message.value), Acc);
  +      #kpro_message{attributes = Attr} = Msg when ?KPRO_IS_LZ4_ATTR(Attr) ->
  +        decode_message_stream(lz4_unpack(Msg#kpro_message.value), Acc);
         _Else ->
           [Msg | Acc]
       end,
  
positive.code:   | encode_messages(KvList)
    ];
  encode_messages([{K, V} | KvList]) ->
    Msg = #kpro_Message{ attributes = ?KPRO_COMPRESS_NONE
                       , key        = K
                       , value      = V
                       },
  
negative.code:                  snappy -> ?KPRO_COMPRESS_SNAPPY;
                   lz4    -> ?KPRO_COMPRESS_LZ4
                 end,
    Msg = #kpro_message{ attributes = Attributes
                       , key        = <<>>
                       , value      = do_compress(Method, IoData)
                       },
  