anchor.diff:  %%%
  -%%%   Copyright (c) 2014, 2015, Klarna AB
  +%%%   Copyright (c) 2014-2016, Klarna AB
   %%%
   %%%   Licensed under the Apache License, Version 2.0 (the "License");
   %%%   you may not use this file except in compliance with the License.
  
positive.code:   end.
  
  connect(Topic, Partition, #state{client_id = ClientId, debug = Dbg} = State) ->
    case fetch_leader_broker_metadata(Topic, Partition, State) of
      {ok, #broker_metadata{host = H, port = P, node_id = LeaderId}} ->
        {ok, Pid} = brod_sock:start_link(self(), H, P, ClientId, Dbg),
        Socket = #socket{pid = Pid, host = H, port = P, node_id = LeaderId},
        Sockets = [Socket | State#state.sockets],
        Leaders = [{{Topic, Partition}, Pid} | State#state.leaders],
        {ok, Pid, State#state{sockets = Sockets, leaders = Leaders}};
      {error, Reason} ->
        {error, Reason}
    end.
  
  -spec fetch_leader_broker_metadata(topic(), partition(), #state{}) ->
          {ok, #broker_metadata{}} | {error, any()}.
  fetch_leader_broker_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
  
negative.code: 
  %%%=============================================================================
  %%% @doc
  %%% @copyright 2014-2016 Klarna AB
  %%% @end
  %%%=============================================================================
  
  