anchor.diff:                         ?MAX_SUP_RESTART_COOLDOWN_MS),
     timer:sleep(SleepMs),
     error_logger:info_msg("Workers supervisor ~p crashed, recovering~n", [Pid]),
  -  ets:delete_all_objects(?T_LEADERS),
  +  TabLeaders = State#state.t_leaders,
  +  ets:delete_all_objects(TabLeaders),
     {ok, SupPid} = brod_producer_sup:start_link(),
  -  {ok, Brokers} = refresh_metadata(State#state.hosts),
  +  {ok, Brokers} = refresh_metadata(State#state.hosts, TabLeaders),
     BlockedList = queue:to_list(State#state.blocked_requests),
     %% loop over ?T_REQUESTS and create new leader queues
     %% skip requests which are in blocked_reqeusts queue
  
positive.code:   Workers =
      lists:keystore(LeaderId, 1, State0#state.workers, {LeaderId, WorkerPid}),
    State = State0#state{workers = Workers},
    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(LeaderId, WorkerPid, EmptyQueue, Blocked0),
        LeaderQueues = [{LeaderId, Queue} | State#state.leader_queues],
        {reply, ok, State#state{ leader_queues = LeaderQueues
                               , blocked_requests = Blocked
  
negative.code:       {error, Error}
    end.
  
  -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(TabLeaders, '_'),
    %% fill in {Topic, Partition} -> Leader mapping table
    [ begin
        Broker = lists:keyfind(Id, #broker_metadata.node_id, Brokers),
        ets:insert(TabLeaders, {{T, P}, Broker})
      end || #topic_metadata{name = T, partitions = PS} <- Topics,
             #partition_metadata{leader_id = Id, id = P} <- PS
    ],
  