anchor.diff:                 , debug            :: [term()]
                  , sup_pid          :: pid()
                  , options          :: proplists:proplist()
  +               , t_leaders        :: ets:tab()
                  , blocked_requests :: queue:queue()
                  , brokers          :: [#broker_metadata{}]
                  , workers          :: [{leader_id(), pid()}]
  
positive.code:   ok = brod_producer_worker:produce(WorkerPid, Ref),
    resend_outstanding(WorkerPid, Tail).
  
  maybe_unblock(LeaderId, WorkerPid, LeaderQueue, BlockedQueue) ->
    L = queue:to_list(BlockedQueue),
    maybe_unblock(LeaderId, WorkerPid, LeaderQueue, L, []).
  
  maybe_unblock(_LeaderId, _WorkerPid, LeaderQueue, [], Acc) ->
    {LeaderQueue, queue:from_list(lists:reverse(Acc))};
  maybe_unblock(LeaderId, WorkerPid, LeaderQueue, [{Ref, From} | Tail], Acc) ->
    [[Topic, Partition]] = ets:match(?T_REQUESTS, #request{ ref = Ref
                                                          , topic = '$1'
                                                          , partition = '$2'
                                                          , _ = '_'}),
    case get_leader_id(Topic, Partition) of
      {ok, LeaderId} ->
        ok = brod_producer_worker:produce(WorkerPid, Ref),
        gen_server:reply(From, {ok, Ref}),
        maybe_unblock(LeaderId, WorkerPid, queue:in(Ref, LeaderQueue), Tail, Acc);
      _ ->
        maybe_unblock(LeaderId, WorkerPid, Tail, [Ref | Acc])
    end.
  
  get_leader_id(Topic, Partition) ->
    case ets:lookup(?T_LEADERS, {Topic, Partition}) of
      [{_, #broker_metadata{node_id = Id}}] ->
        {ok, Id};
      [] ->
  
negative.code:             false ->
                Topic = R#request.topic,
                Partition = R#request.partition,
                {ok, LeaderId} = get_leader_id(TabLeaders, Topic, Partition),
                case lists:keyfind(LeaderId, 1, LeaderQueues) of
                  {_, Q} ->
                    lists:keystore(LeaderId, 1, LeaderQueues,
  