anchor.diff:    #broker_metadata{host = Host, port = Port} =
       lists:keyfind(LeaderId, #broker_metadata.node_id, Brokers),
   
  -  gen_server:call(ClientId, {get_connection, Host, Port}, infinity).
  +  get_connection(ClientId, Host, Port).
   
   -spec get_metadata(client_id(), topic()) -> {ok, #metadata_response{}}.
   get_metadata(Client, Topic) ->
     gen_server:call(Client, {get_metadata, Topic}, infinity).
  +
  +%% @doc Get the connection to kafka broker at Host:Port.
  +%% If there is no such connection established yet, try to establish a new one.
  +%% If the connection is already established per request from another producer
  +%% the same socket is returned.
  +%% If the old connection was dead less than a configurable N seconds ago,
  +%% {error, LastReason} is returned.
  +%% @end
  +-spec get_connection(client(), hostname(), portnum()) ->
  +                        {ok, pid()} | {error, any()}.
  +get_connection(Client, Host, Port) ->
  +  gen_server:call(Client, {get_connection, Host, Port}, infinity).
   
   %% @doc Get all partition numbers of a given topic.
   -spec get_partitions(client(), topic()) -> {ok, [partition()]} | {error, any()}.
  
positive.code: stop(Client) ->
    gen_server:call(Client, stop, infinity).
  
  %% @doc Get the connection to kafka broker at Host:Port which is a leader
  %% for Topic.Partition.
  %% If there is no such connection established yet, try to establish a new one.
  %% If the connection is already established per request from another
  %% producer/consumer the same socket is returned.
  
negative.code: 
  -export([ find_producer/3
          , find_consumer/3
          , get_connection/3
          , get_leader_connection/3
          , get_metadata/2
          , get_partitions/2
  