anchor.diff:                          Port:32/integer,
                           Bin/binary>>) ->
     Broker = #broker_metadata{ node_id = NodeID
  -                           , host = binary_to_list(Host)
  +                           , host = binary_to_list(binary:copy(Host))
                              , port = Port},
     {Broker, 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:                        Bin0/binary>>) ->
    {Partitions, Bin} = parse_array(Bin0, fun parse_partition_metadata/1),
    Topic = #topic_metadata{ error_code = ErrorCode
                           , name = binary:copy(Name)
                           , partitions = Partitions},
    {Topic, Bin}.
  
  