anchor.diff:    parse_array(Length, Bin, [], Fun).
   
   parse_array(0, Bin, Acc, _Fun) ->
  -  {lists:reverse(Acc), Bin};
  +  {Acc, Bin};
   parse_array(Length, Bin0, Acc, Fun) ->
     {Item, Bin} = Fun(Bin0),
     parse_array(Length - 1, Bin, [Item | Acc], Fun).
   
  -parse_int32(<<X:32/integer, Bin/binary>>) -> {X, Bin}.
  -
  -parse_int64(<<X:64/integer, Bin/binary>>) -> {X, Bin}.
  +parse_int32(<<X:32/integer, Bin/binary>>) -> {X, binary:copy(Bin)}.
  +
  +parse_int64(<<X:64/integer, Bin/binary>>) -> {X, binary:copy(Bin)}.
   
   kafka_size(<<"">>) -> -1;
   kafka_size(Bin)    -> size(Bin).
  
positive.code:   Timeout = Produce#produce_request.timeout,
    Topics = group_by_topic(Produce#produce_request.data, []),
    TopicsCount = erlang:length(Topics),
    TopicsBin = encode_topics(Topics, <<>>),
    <<Acks:16/signed-integer, Timeout:32/integer,
      TopicsCount:32/integer, TopicsBin/binary>>.
  
  group_by_topic([], Acc) ->
    Acc;
  
negative.code: encode_partitions([{Id, Messages} | T], Acc0) ->
    MessageSet = encode_message_set(Messages),
    MessageSetSize = erlang:size(MessageSet),
    Acc = <<Acc0/binary,
            Id:32/integer,
            MessageSetSize:32/integer,
            MessageSet/binary>>,
    encode_partitions(T, Acc).
  
  encode_message_set(Messages) ->
    encode_message_set(Messages, <<>>).
  
  encode_message_set([], Acc) ->
    Acc;
  encode_message_set([Msg | Messages], Acc0) ->
    MsgBin = encode_message(Msg),
    Size = size(MsgBin),
    %% 0 is for offset which is unknown until message is handled by
    %% server, we can put any number here actually
    Acc = <<Acc0/binary, 0:64/integer, Size:32/integer, MsgBin/binary>>,
    encode_message_set(Messages, Acc).
  
  encode_message({Key, Value}) ->
    KeySize = kafka_size(Key),
    ValSize = kafka_size(Value),
    Message = <<?MAGIC_BYTE:8/integer,
                ?COMPRESS_NONE:8/integer,
                KeySize:32/signed-integer,
                Key/binary,
                ValSize:32/signed-integer,
                Value/binary>>,
    Crc32 = erlang:crc32(Message),
    <<Crc32:32/integer, Message/binary>>.
  
  