anchor.diff:        erlang:error({E, Context, erlang:get_stacktrace()})
       end,
     Result =
  -    #kpro_Response{ correlationId   = CorrId
  -                  , responseMessage = Message
  +    #kpro_response{ header = #kpro_response_header{ correlation_id = CorrId }
  +                  , message = Message
                     },
     {Result, Rest};
   do_decode_response( Bin) ->
  
positive.code: next_corr_id(CorrId)       -> CorrId + 1.
  
  %% @doc Parse binary stream received from kafka broker.
  %%      Return a list of kpro_Response() and the remaining bytes.
  %% @end
  -spec decode_response(binary()) -> {[kpro_Response()], binary()}.
  decode_response(Bin) ->
    decode_response(Bin, []).
  
  
negative.code: %% @end
  -spec produce_request(topic(), partition(), kv_list(),
                        integer(), non_neg_integer(),
                        kpro_compress_option()) -> kpro_produce_request_v0().
  produce_request(Topic, Partition, KvList,
                  RequiredAcks, AckTimeout, CompressOption) ->
    Messages = encode_messages(KvList, CompressOption),
    PartitionMsgSet =
      #kpro_produce_request_v0_data{ partition  = Partition
                                   , record_set = Messages
                                   },
    TopicMessageSet =
      #kpro_produce_request_v0_topic_data{ topic = Topic
                                         , data  = [PartitionMsgSet]
                                         },
    %% Encode message set here right now.
    %% Instead of keeping a possibily very large array
    %% and passing it around processes
    %% e.g. in brod, the message set can be encoded in producer
    %% worker before sending it down to socket process
    MessageSetBin = iolist_to_binary(encode({array, [TopicMessageSet]})),
    #kpro_produce_request_v0{ acks       = RequiredAcks
                            , timeout    = AckTimeout
                            , topic_data = {already_encoded, MessageSetBin}
                            }.
  
  %% @doc Get the next correlation ID.
  -spec next_corr_id(corr_id()) -> corr_id().
  