anchor.diff:               , topic/0
                , partition/0
                , offset/0
  +             , key/0
  +             , value/0
                , kafka_key/0
                , kafka_value/0
  +             , corr_id/0
                ]).
   
   -include("kpro.hrl").
  
positive.code: do_to_maps(KVL) -> KVL.
  -endif.
  
  messages(KafkaKvList, Compression) ->
    Messages =
      lists:map(fun({K, V}) ->
                  #kpro_Message{ attributes = ?KPRO_COMPRESS_NONE
                               , key        = K
                               , value      = V
                               }
                end, KafkaKvList),
    case Compression =:= no_compression of
      true  -> Messages;
      false -> compress(Compression, Messages)
    end.
  
  compress(Method, Messages) ->
    IoData = [encode(Message) || Message <- Messages],
    Attributes = case Method of
                   gzip   -> ?KPRO_COMPRESS_GZIP;
                   snappy -> ?KPRO_COMPRESS_SNAPPY;
  
negative.code: -type topic()     :: str().
  -type partition() :: int32().
  -type offset()    :: int64().
  
  -type key() :: undefined | iodata().
  -type value() :: undefined | iodata() | [{key(), kv_list()}].
  -type kv_list() :: [{key(), value()}].
  
  -type kafka_key() :: key().
  -type kafka_value() :: undefined | iodata() | [kpro_Message()].
  
  -define(INT, signed-integer).
  
  