anchor.diff:  -type kv_list() :: [{key(), value()}].
   
   -type kafka_key() :: key().
  --type kafka_value() :: undefined | iodata() | [kpro_Message()].
  +-type kafka_value() :: undefined | iodata() | [kpro_message()].
   
   -type incomplete_message() :: {?incomplete_message, int32()}.
   
  
positive.code:                  snappy -> ?KPRO_COMPRESS_SNAPPY;
                   lz4    -> ?KPRO_COMPRESS_LZ4
                 end,
    Msg = #kpro_Message{ attributes = Attributes
                       , key        = <<>>
                       , value      = do_compress(Method, IoData)
                       },
  
negative.code: 
  %% @private
  -spec decode_message(binary()) ->
          {kpro_message() | incomplete_message(), binary()}.
  decode_message(<<_Offset:64/?INT, MsgSize:32/?INT, T/binary>> = Bin) ->
    case size(T) < MsgSize of
      true  -> {{?incomplete_message, MsgSize + 12}, <<>>};
      false -> decode(kpro_message, Bin)
    end;
  decode_message(_) ->
    %% need to fetch at least 12 bytes to know the message size
  