anchor.diff:        end
     end.
   
  --spec lookup_partitions_count_cache(ets:tab(), ?undef | topic()) ->
  +-spec lookup_partitions_count_cache(ets:tab(), ?undef | brod:topic()) ->
           {ok, pos_integer()} | {error, any()} | false.
   lookup_partitions_count_cache(_Ets, ?undef) -> false;
   lookup_partitions_count_cache(Ets, Topic) ->
  
positive.code:   end.
  
  %% @doc Get topic metadata, if topic is 'undefined', will fetch ALL metadata.
  -spec get_metadata(client(), ?undef | topic()) ->
          {ok, kpro_MetadataResponse()} | {error, any()}.
  get_metadata(Client, Topic) ->
    safe_gen_call(Client, {get_metadata, Topic}, infinity).
  
  %% @doc Get number of partitions for a given topic.
  -spec get_partitions_count(client(), topic()) ->
          {ok, pos_integer()} | {error, any()}.
  get_partitions_count(Client, Topic) when is_atom(Client) ->
    %% Ets =:= ClientId
  
negative.code: 
  -define(UNKNOWN_TOPIC_CACHE_EXPIRE_SECONDS, 120).
  
  -type partition_worker_key() :: ?PRODUCER_KEY(brod:topic(), brod:partition())
                                | ?CONSUMER_KEY(brod:topic(), brod:partition()).
  
  -type get_producer_error() :: client_down
                              | {producer_down, noproc}
                              | {producer_not_found, brod:topic()}
                              | {producer_not_found, brod:topic(), brod:partition()}.
  
  -type get_consumer_error() :: client_down
                              | {consumer_down, noproc}
                              | {consumer_not_found, brod:topic()}
                              | {consumer_not_found, brod:topic(), brod:partition()}.
  
  -type get_worker_error() :: get_producer_error()
                            | get_consumer_error().
  
  -record(sock,
          { endpoint :: brod:endpoint()
          , sock_pid :: pid() | dead_socket()
          }).
  
  -record(state,
          { client_id            :: brod:brod_client_id()
          , bootstrap_endpoints  :: [brod:endpoint()]
          , meta_sock_pid        :: pid() | dead_socket()
          , payload_sockets = [] :: [#sock{}]
          , producers_sup        :: pid()
          , consumers_sup        :: pid()
          , config               :: brod:client_config()
          , workers_tab          :: ets:tab()
          }).
  
  %%%_* APIs =====================================================================
  
  -spec start_link( [brod:endpoint()]
                  , brod:brod_client_id()
                  , brod:client_config()) -> {ok, pid()} | {error, any()}.
  start_link(BootstrapEndpoints, ClientId, Config) when is_atom(ClientId) ->
    Args = {BootstrapEndpoints, ClientId, Config},
    gen_server:start_link({local, ClientId}, ?MODULE, Args, []).
  
  -spec stop(brod:client()) -> ok.
  stop(Client) ->
    Mref = erlang:monitor(process, Client),
    _ = safe_gen_call(Client, stop, infinity),
  