anchor.diff:    true = (CorrId0 =< ?MAX_CORR_ID), %% assert
     ApiKey = req_to_api_key(RequestMessage),
     CorrId = (ApiKey bsl ?CORR_ID_BITS) bor CorrId0,
  -  ApiVersion = case ApiVersion0 =:= undefined of
  -                  true  -> get_api_version(RequestMessage);
  -                  false -> ApiVersion0
  -               end,
  +  ApiVersion = get_api_version(ApiVersion0, RequestMessage),
     IoData =
       [ encode({int16, ApiKey})
       , encode({int16, ApiVersion})
  
positive.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}),
  
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).
  
  