anchor.diff:                           [ClientId, Pid, Reason]),
     {stop, {consumers_sup_down, Reason}, State};
   handle_info({'EXIT', Pid, Reason},
  -            #state{ client_id = ClientId
  -                  , meta_sock = #sock{sock_pid = Pid}
  +            #state{ client_id           = ClientId
  +                  , meta_sock_pid       = Pid
  +                  , bootstrap_endpoints = BootstrapEndpoints0
                     } = State) ->
  -  MetaSock = State#state.meta_sock,
  -  {Host, Port} = MetaSock#sock.endpoint,
  +  [{Host, Port} | Rest] = BootstrapEndpoints0,
     error_logger:error_msg("client ~p metadata socket down ~s:~p~nReason:~p",
                           [ClientId, Host, Port, Reason]),
  -  {ok, NewMetaSock} = mark_socket_dead(MetaSock, Reason),
  -  {noreply, State#state{meta_sock = NewMetaSock}};
  +  %% move the newly failed endpoint to the last in the list
  +  BootstrapEndpoints = Rest ++ [{Host, Port}],
  +  NewState =
  +    State#state{ meta_sock_pid       = ?dead_since(os:timestamp(), Reason)
  +               , bootstrap_endpoints = BootstrapEndpoints
  +               },
  +  {noreply, NewState};
   handle_info({'EXIT', Pid, Reason}, State) ->
     {ok, NewState} = handle_socket_down(State, Pid, Reason),
     {noreply, NewState};
  
positive.code:   {noreply, State}.
  
  handle_call({start_producer, TopicName, ProducerConfig}, _From, State) ->
    try
      {ok, MetaSock} = get_meta_sock(State),
      is_topic_exist(MetaSock#sock.sock_pid, TopicName) orelse
        throw({invalid_topic, TopicName}),
      SupPid = State#state.producers_sup,
      case brod_producers_sup:start_producer(SupPid, self(),
                                             TopicName, ProducerConfig) of
        {ok, _} -> {reply, ok, State#state{meta_sock = MetaSock}};
        Error   -> throw(Error)
      end
    catch throw:E ->
        {reply, E, State}
    end;
  handle_call({start_consumer, TopicName, ConsumerConfig}, _From, State) ->
    try
      {ok, MetaSock} = get_meta_sock(State),
      is_topic_exist(MetaSock#sock.sock_pid, TopicName) orelse
        throw({invalid_topic, TopicName}),
      SupPid = State#state.consumers_sup,
      case brod_consumers_sup:start_consumer(SupPid, self(),
                                             TopicName, ConsumerConfig) of
        {ok, _} -> {reply, ok, State#state{meta_sock = MetaSock}};
        Error   -> throw(Error)
      end
    catch throw:E ->
        {reply, E, State}
    end;
  handle_call(get_workers_table, _From, State) ->
    {reply, State#state.workers_tab, State};
  
negative.code:               Partition
              end, Partitions).
  
  -spec validate_topic_existence(topic(), #state{}) -> {Result, #state{}}
          when Result :: ok | {error, any()}.
  validate_topic_existence(Topic0, #state{config = Config} = State) ->
    %% by default, brod respects what is configured in broker about
    %% topic auto-creation. i.e. whatever auto.create.topics.enable
    %% is set in borker configuration.
    %% however if 'allow_topic_auto_creation' is set to false in client
    %% config, brod will avoid sending metadata requests that may cause an
    %% auto-creation of the topic.
    Topic =
      case proplists:get_value(allow_topic_auto_creation, Config, true) of
        true  -> Topic0;
        false -> ?undef %% ?undef means fetch all topics
      end,
    case do_get_metadata(Topic, State) of
      {{ok, Response}, NewState} ->
        #kpro_MetadataResponse{topicMetadata_L = Topics} = Response,
        case lists:keyfind(Topic0, #kpro_TopicMetadata.topicName, Topics) of
          #kpro_TopicMetadata{} -> {ok, NewState};
          false                 -> {error, topic_not_found}
        end;
      {{error, Reason}, NewState} ->
        {{error, Reason}, NewState}
    end.
  
  -spec do_get_metadata(?undef | topic(), #state{}) -> {Result, #state{}}
          when Result :: {ok, kpro_MetadataResponse()} | {error, any()}.
  do_get_metadata(Topic, #state{config = Config} = State) ->
    {ok, NewState} = maybe_restart_metadata_socket(State),
    Topics = case Topic of
               ?undef -> []; %% in case no topic is given, get all
               _      -> [Topic]
             end,
    Request = #kpro_MetadataRequest{topicName_L = Topics},
    Timeout = proplists:get_value(get_metadata_timout_seconds, Config,
                                  ?DEFAULT_GET_METADATA_TIMEOUT_SECONDS),
    SockPid = NewState#state.meta_sock_pid,
    Result = brod_sock:send_sync(SockPid, Request, timer:seconds(Timeout)),
    {Result, NewState}.
  
  -spec do_get_connection(#state{}, hostname(), portnum()) ->
          {#state{}, Result} when Result :: {ok, pid()} | {error, any()}.
  