anchor.diff:    end.
   
   connect(Topic, Partition, 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, State#state.debug),
  +      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
  
positive.code:           {error, {unknown_partition, Topic, Partition}};
          #partition_metadata{leader_id = LeaderId} ->
            Broker = lists:keyfind(LeaderId, #broker_metadata.node_id, Brokers),
            #broker_metadata{host = H, port = P} = Broker,
            {ok, Pid} = brod_sock:start_link(self(), H, P, State#state.debug),
            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}}
        end
    end.
  
  
negative.code: -type topic()     :: binary().
  -type partition() :: non_neg_integer().
  -type leader()    :: pid(). % brod_sock pid
  -type corr_id()   :: 0..2147483647.
  -type sender()    :: {pid(), reference()}.
  -type offset()    :: integer().
  
  -ifdef(otp_before_17).
  -type data_buffer() :: [{leader(), [{topic(), dict()}]}].
  