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: code_change(_OldVsn, State, _Extra) ->
    {ok, State}.
  
  terminate(Reason, #state{ client_id     = ClientId
                          , meta_sock     = MetaSock
                          , sockets       = Sockets
                          , producers_sup = ProducersSup
                          , consumers_sup = ConsumersSup
                          }) ->
    case brod_utils:is_normal_reason(Reason) of
      true ->
  
negative.code: 
  -spec connect(#state{}, hostname(), portnum()) -> {#state{}, Result}
          when Result :: {ok, pid()} | {error, any()}.
  connect(#state{ client_id       = ClientId
                , payload_sockets = Sockets
                } = State, Host, Port) ->
    Endpoint = {Host, Port},
    case brod_sock:start_link(self(), Host, Port, ClientId, []) of
  