anchor.diff:        error_logger:info_msg("client ~p connected to ~s:~p~n",
                               [ClientId, Host, Port]),
         NewSockets = lists:keystore(Endpoint, #sock.endpoint, Sockets, S),
  -      {State#state{sockets = NewSockets}, {ok, Pid}};
  +      {State#state{payload_sockets = NewSockets}, {ok, Pid}};
       {error, Reason} ->
         error_logger:error_msg("client ~p failed to connect to ~s:~p~n"
                                "reason:~p",
                                [ClientId, Host, Port, Reason]),
         {ok, Sock} = mark_socket_dead(#sock{endpoint = Endpoint}, Reason),
         NewSockets = lists:keystore(Endpoint, #sock.endpoint, Sockets, Sock),
  -      {State#state{sockets = NewSockets}, {error, Reason}}
  +      {State#state{payload_sockets = NewSockets}, {error, Reason}}
     end.
   
   %% @private Handle socket pid EXIT event, keep the timestamp.
  
positive.code:         }).
  
  -record(state,
          { client_id     :: client_id()
          , endpoints     :: [endpoint()]
          , meta_sock     :: pid()
          , sockets = []  :: [#sock{}]
          , producers_sup :: pid()
          , consumers_sup :: pid()
          , config        :: client_config()
          , workers_tab   :: ets:tab()
          }).
  
  %%%_* APIs =====================================================================
  
  -spec start_link( [endpoint()]
                  , client_config()) -> {ok, pid()} | {error, any()}.
  start_link(Endpoints, Config) ->
    Args = {Endpoints, Config, [], []},
    gen_server:start_link(?MODULE, Args, []).
  
  -spec start_link( [endpoint()]
                  , client_id()
                  , client_config()) -> {ok, pid()} | {error, any()}.
  start_link(Endpoints, ClientId, Config) ->
    start_link(Endpoints, ClientId, Config, [], []).
  
  -spec start_link( [endpoint()]
                  , client_id()
  
negative.code:   Diff = timer:now_diff(Now, Ts) div 1000000,
    Diff >= Threshold.
  
  %% @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.
  