anchor.diff:                       , key        = <<>>
                        , value      = do_compress(Method, IoData)
                        },
  -  [Msg].
  +  [encode(Msg)].
  +
   
   %% TODO: add snappy and lz4 compression
  +-spec do_compress(kpro_compress_option(), iodata()) -> iodata().
   do_compress(gzip, IoData) ->
     zlib:gzip(IoData).
   
  -get_api_version(#kpro_OffsetCommitRequestV1{}) -> 1;
  -get_api_version(#kpro_OffsetCommitRequestV2{}) -> 2;
  -get_api_version(#kpro_OffsetFetchRequest{})    -> 1;
  -get_api_version(_)                             -> 0.
  +-spec get_api_version(int16() | undefined, kpro_RequestMessage()) -> int16().
  +get_api_version(V, _Msg) when is_integer(V) -> V;
  +get_api_version(undefined, Msg)             -> api_version(Msg).
  +
  +-spec api_version(kpro_RequestMessage()) -> int16().
  +api_version(#kpro_OffsetCommitRequestV1{}) -> 1;
  +api_version(#kpro_OffsetCommitRequestV2{}) -> 2;
  +api_version(#kpro_OffsetFetchRequest{})    -> 1;
  +api_version(_Default)                      -> 0.
   
   %% @private Decode responses received from kafka broker.
   %% {incomplete, TheOriginalBinary} is returned if this is not a complete packet.
  
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:                  undefined            -> ?KPRO_ATTRIBUTES;
                   A when is_integer(A) -> A
                 end,
    Body = [ encode({int8, MagicByte})
           , encode({int8, Attributes})
           , encode({bytes, R#kpro_Message.key})
           , encode({bytes, R#kpro_Message.value})
           ],
    Crc  = encode({int32, erlang:crc32(Body)}),
    Size = data_size([Crc, Body]),
    [encode({int64, -1}),
  