anchor.diff:    Diff = timer:now_diff(Now, Ts) div 1000000,
     Diff >= Threshold.
   
  -%% @private Establish a dedicated socket to kafka cluster endpoint(s) for
  -%%          metadata retrievals.
  +%% @private Maybe restart the metadata socket pid if it is no longer alive.
  +-spec maybe_restart_metadata_socket(#state{}) -> {ok, #state{}} | no_return().
  +maybe_restart_metadata_socket(#state{meta_sock_pid = MetaSockPid} = State) ->
  +  case is_pid(MetaSockPid) andalso is_process_alive(MetaSockPid) of
  +    true ->
  +      {ok, State};
  +    false -> % can happen when metadata connection closed
  +      ClientId = State#state.client_id,
  +      Endpoints = State#state.bootstrap_endpoints,
  +      {ok, NewMetaSockPid, ReorderedEndpoints} =
  +        start_metadata_socket(ClientId, Endpoints),
  +      {ok, State#state{ bootstrap_endpoints = ReorderedEndpoints
  +                      , meta_sock_pid       = NewMetaSockPid
  +                      }}
  +  end.
  +
  +%% @private Establish a dedicated socket to kafka cluster bootstrap endpoint(s)
  +%% for metadata retrievals. Failed endpoints are moved to the end of the list
  +%% so that future retries will start from one that has not tried yet, or the
  +%% one failed longest ago.
  +%%
   %% NOTE: This socket is not intended for kafka payload. This is to avoid
   %%       burst of connection usage when many partition producers (re)start
   %%       at same time, if we use brod_util:get_metadata/2 to fetch metadata.
  
positive.code: 
  %%%_* gen_server callbacks =====================================================
  
  init({Endpoints, ClientId, Config, Producers, Consumers}) ->
    erlang:process_flag(trap_exit, true),
    Tab = ets:new(?ETS(ClientId),
                  [named_table, protected, {read_concurrency, true}]),
    self() ! {init, Producers, Consumers},
    {ok, #state{ client_id   = ClientId
               , endpoints   = Endpoints
               , config      = Config
               , workers_tab = Tab
               }};
  init({Endpoints, Config, Producers, Consumers}) ->
    erlang:process_flag(trap_exit, true),
    Tab = ets:new(workers_tab, [protected, {read_concurrency, true}]),
    self() ! {init, Producers, Consumers},
    {ok, #state{ client_id   = ?DEFAULT_CLIENT_ID
               , endpoints   = Endpoints
               , config      = Config
               , workers_tab = Tab
               }}.
  
  handle_info({init, Producers, Consumers}, State) ->
    ClientId = State#state.client_id,
    Endpoints = State#state.endpoints,
    {ok, MetaSock} = start_metadata_socket(ClientId, Endpoints),
    {ok, ProducersSupPid} = brod_producers_sup:start_link(self(), Producers),
    {ok, ConsumersSupPid} = brod_consumers_sup:start_link(self(), Consumers),
    {noreply, State#state{ meta_sock     = MetaSock
                         , producers_sup = ProducersSupPid
                         , consumers_sup = ConsumersSupPid
                         }};
  
  handle_info({'EXIT', Pid, Reason}, #state{ client_id     = ClientId
  
negative.code:                 , [{topic(), producer_config()}]
                  , [{topic(), consumer_config()}]) ->
                      {ok, pid()} | {error, any()}.
  start_link(BootstrapEndpoints, ClientId, Config, Producers, Consumers)
    when is_atom(ClientId) ->
    Args = {BootstrapEndpoints, ClientId, Config, Producers, Consumers},
    gen_server:start_link({local, ClientId}, ?MODULE, Args, []).
  
  -spec stop(client()) -> ok.
  