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:   [<<Length:32/?INT>>, [encode(I) || I <- L]];
  encode(#kpro_PartitionMessageSet{} = R) ->
    %% messages in messageset is a stream, not an array
    MessageSet = [encode(M) || M <- R#kpro_PartitionMessageSet.message_L],
    Size = data_size(MessageSet),
    [encode({int32, R#kpro_PartitionMessageSet.partition}),
     encode({int32, Size}),
     MessageSet
    ];
  encode(#kpro_Message{} = R) ->
    MagicByte = case R#kpro_Message.magicByte of
  
negative.code: %% @equiv produce_request(Topic, Partition, KvList, RequiredAcks,
  %%                        AckTimeout, no_compression)
  %% @end
  -spec produce_request(topic(), partition(), kv_list(),
                        integer(), non_neg_integer()) ->
                          kpro_ProduceRequest().
  produce_request(Topic, Partition, KvList, RequiredAcks, AckTimeout) ->
    produce_request(Topic, Partition, KvList, RequiredAcks, AckTimeout,
                    no_compression).
  
  %% @doc Help function to construct a #kpro_ProduceRequest{} for
  %% messages targeting one single topic-partition.
  %% @end
  -spec produce_request(topic(), partition(), kv_list(),
                        integer(), non_neg_integer(),
                        kpro_compress_option()) -> kpro_ProduceRequest().
  produce_request(Topic, Partition, KvList,
                  RequiredAcks, AckTimeout, CompressOption) ->
    Messages = encode_messages(KvList, CompressOption),
    PartitionMsgSet =
      #kpro_PartitionMessageSet{ partition = Partition
                               , message_L = Messages
  