anchor.diff:  
   %%%=============================================================================
   %%% @doc
  -%%% @copyright 2014, 2015 Klarna AB
  +%%% @copyright 2014-2016 Klarna AB
   %%% @end
   %%%=============================================================================
   
  
positive.code: %%%
  %%%   Copyright (c) 2014, 2015, Klarna AB
  %%%
  %%%   Licensed under the Apache License, Version 2.0 (the "License");
  %%%   you may not use this file except in compliance with the License.
  
negative.code:   end.
  
  connect(Topic, Partition, #state{client_id = ClientId, debug = Dbg} = State) ->
    case fetch_metadata(Topic, Partition, State) of
      {ok, Leader, LeaderPartitions} ->
        #broker_metadata{host = Host, port = Port, node_id = LeaderId} = Leader,
        {ok, Pid} = brod_sock:start_link(self(), Host, Port, ClientId, Dbg),
        Socket = #socket{pid = Pid, host = Host, port = Port, node_id = LeaderId},
        Sockets = [Socket | State#state.sockets],
        NewLeader = lists:map(fun({T, P}) -> {{T, P}, Pid} end, LeaderPartitions),
        Leaders = lists:append(NewLeader, State#state.leaders),
        {ok, Pid, State#state{sockets = Sockets, leaders = Leaders}};
      {error, Reason} ->
        {error, Reason}
    end.
  
  -spec fetch_metadata(topic(), partition(), #state{}) ->
          {ok, #broker_metadata{}} | {error, any()}.
  fetch_metadata(Topic, Partition, State) ->
    {ok, Metadata} = brod_utils:get_metadata(State#state.hosts),
    #metadata_response{brokers = Brokers, topics = Topics} = Metadata,
    case lists:keyfind(Topic, #topic_metadata.name, Topics) of
  