anchor.diff:        {{error, Reason}, NewState}
     end.
   
  --spec do_get_metadata(?undef | topic(), #state{}) -> {Result, #state{}}
  +-spec do_get_metadata(?undef | brod:topic(), #state{}) -> {Result, #state{}}
           when Result :: {ok, kpro_MetadataResponse()} | {error, any()}.
   do_get_metadata(Topic, #state{ client_id   = ClientId
                                , config      = Config
  
positive.code:   end.
  
  %% @doc Get consumer of the given topic-parition.
  -spec get_consumer(client(), topic(), partition()) ->
          {ok, pid()} | {error, get_consumer_error()}.
  get_consumer(Client, Topic, Partition) ->
    get_partition_worker(Client, ?CONSUMER_KEY(Topic, Partition)).
  
negative.code:   end.
  
  %% @doc Stop all partition consumers of the given topic.
  -spec stop_consumer(brod:client(), brod:topic()) -> ok | {error, any()}.
  stop_consumer(Client, TopicName) ->
    safe_gen_call(Client, {stop_consumer, TopicName}, infinity).
  
  