anchor.diff:  encode_partitions([{Id, Messages} | T], Acc0) ->
     MessageSet = encode_message_set(Messages),
     MessageSetSize = erlang:size(MessageSet),
  -  Acc = <<Id:32/integer,
  +  Acc = <<Acc0/binary,
  +          Id:32/integer,
             MessageSetSize:32/integer,
  -          MessageSet/binary,
  -          Acc0/binary>>,
  +          MessageSet/binary>>,
     encode_partitions(T, Acc).
   
   encode_message_set(Messages) ->
  -  F = fun(X, Acc) ->
  -          Message = encode_message(X),
  -          Size = size(Message),
  -          %% 0 is for offset which is unknown until message is handled by
  -          %% server, we can put any number here actually
  -          <<0:64/integer, Size:32/integer, Message/binary, Acc/binary>>
  -      end,
  -  MessageSet = lists:foldr(F, <<>>, Messages),
  -  <<MessageSet/binary>>.
  +  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),
  -  Payload = <<KeySize:32/signed-integer,
  +  Message = <<?MAGIC_BYTE:8/integer,
  +              ?COMPRESS_NONE:8/integer,
  +              KeySize:32/signed-integer,
                 Key/binary,
                 ValSize:32/signed-integer,
                 Value/binary>>,
  -  Message = <<?MAGIC_BYTE:8/integer, ?COMPRESS_NONE:8/integer, Payload/binary>>,
     Crc32 = erlang:crc32(Message),
     <<Crc32:32/integer, Message/binary>>.
   
  
positive.code: 
  parse_produce_topic(<<Size:16/integer, Name:Size/binary, Bin0/binary>>) ->
    {Offsets, Bin} = parse_array(Bin0, fun parse_produce_offset/1),
    {#produce_topic{topic = Name, offsets = Offsets}, Bin}.
  
  parse_produce_offset(<<Partition:32/integer,
                         ErrorCode:16/signed-integer,
  
negative.code: 
  parse_offset_topic(<<Size:16/integer, Name:Size/binary, Bin0/binary>>) ->
    {Partitions, Bin} = parse_array(Bin0, fun parse_partition_offsets/1),
    {#offset_topic{topic = binary:copy(Name), partitions = Partitions}, Bin}.
  
  parse_partition_offsets(<<Partition:32/integer,
                            ErrorCode:16/signed-integer,
  