anchor.diff:    end.
   
   spawn_worker(Topic, Partition, State) ->
  -  case ets:lookup(?T_LEADERS, {Topic, Partition}) of
  +  case ets:lookup(State#state.t_leaders, {Topic, Partition}) of
       [{_, #broker_metadata{} = Broker}] ->
         spawn_worker(Broker, State);
       [] ->
  
positive.code:       {error, Error}
    end.
  
  -spec refresh_metadata([brod:host()]) -> ok.
  refresh_metadata(Hosts) ->
    {ok, Metadata} = brod_utils:get_metadata(Hosts),
    #metadata_response{brokers = Brokers, topics = Topics} = Metadata,
    ets:match_delete(?T_LEADERS, '_'),
    %% fill in {Topic, Partition} -> Leader mapping table
    [ begin
        Broker = lists:keyfind(Id, #broker_metadata.node_id, Brokers),
        ets:insert(?T_LEADERS, {{T, P}, Broker})
      end || #topic_metadata{name = T, partitions = PS} <- Topics,
             #partition_metadata{leader_id = Id, id = P} <- PS
    ],
  
negative.code:   erlang:process_flag(trap_exit, true),
    MaxLeaderQueueLen = proplists:get_value(max_leader_queue_len, Options,
                                            ?DEFAULT_MAX_LEADER_QUEUE_LEN),
    TabLeaders = ets:new(t_leaders, [protected]),
    ets:new(?T_REQUESTS, [protected, named_table, ordered_set,
                          {keypos, #request.ref}]),
    {ok, SupPid} = brod_producer_sup:start_link(),
    {ok, Brokers} = refresh_metadata(Hosts, TabLeaders),
    {ok, #state{ hosts            = Hosts
               , debug            = Debug
               , sup_pid          = SupPid
               , options          = Options
               , t_leaders        = TabLeaders
               , blocked_requests = queue:new()
               , brokers          = Brokers
               , workers          = []
  