anchor.diff:        {error, Error}
     end.
   
  --spec refresh_metadata([brod:host()]) -> ok.
  -refresh_metadata(Hosts) ->
  +-spec refresh_metadata([brod:host()], ets:tab()) -> ok.
  +refresh_metadata(Hosts, TabLeaders) ->
     {ok, Metadata} = brod_utils:get_metadata(Hosts),
     #metadata_response{brokers = Brokers, topics = Topics} = Metadata,
  -  ets:match_delete(?T_LEADERS, '_'),
  +  ets:match_delete(TabLeaders, '_'),
     %% fill in {Topic, Partition} -> Leader mapping table
     [ begin
         Broker = lists:keyfind(Id, #broker_metadata.node_id, Brokers),
  -      ets:insert(?T_LEADERS, {{T, P}, Broker})
  +      ets:insert(TabLeaders, {{T, P}, Broker})
       end || #topic_metadata{name = T, partitions = PS} <- Topics,
              #partition_metadata{leader_id = Id, id = P} <- PS
     ],
  
positive.code:   end.
  
  spawn_worker(Topic, Partition, State) ->
    case ets:lookup(?T_LEADERS, {Topic, Partition}) of
      [{_, #broker_metadata{} = Broker}] ->
        spawn_worker(Broker, State);
      [] ->
  
negative.code:   Workers =
      lists:keystore(LeaderId, 1, State0#state.workers, {LeaderId, WorkerPid}),
    State = State0#state{workers = Workers},
    TabLeaders = State#state.t_leaders,
    EmptyQueue = queue:new(),
    case lists:keyfind(LeaderId, 1, State#state.leader_queues) of
      Res when Res =:= false orelse Res =:= EmptyQueue ->
        %% try to unblock producers when there are no pending requests
        Blocked0 = State#state.blocked_requests,
        {Queue, Blocked} =
          maybe_unblock(TabLeaders, LeaderId, WorkerPid, EmptyQueue, Blocked0),
        LeaderQueues = [{LeaderId, Queue} | State#state.leader_queues],
        {reply, ok, State#state{ leader_queues = LeaderQueues
                               , blocked_requests = Blocked
  