anchor.diff:    end.
   
   %% @doc Stop all partition producers of the given topic.
  --spec stop_producer(client(), topic()) -> ok | {error, any()}.
  +-spec stop_producer(brod:client(), brod:topic()) -> ok | {error, any()}.
   stop_producer(Client, TopicName) ->
     safe_gen_call(Client, {stop_producer, TopicName}, infinity).
   
   %% @doc Dynamically start a topic consumer.
   %% Returns ok if the consumer is already started.
   %% @end.
  --spec start_consumer(client(), topic(), consumer_config()) ->
  +-spec start_consumer(brod:client(), brod:topic(), brod:consumer_config()) ->
                           ok | {error, any()}.
   start_consumer(Client, TopicName, ConsumerConfig) ->
     case get_consumer(Client, TopicName, _Partition = 0) of
  
positive.code:   end.
  
  
  -spec maybe_connect(#state{}, hostname(), portnum(), Reason) ->
          {#state{}, Result} when
            Reason :: not_found | dead_socket(),
            Result :: {ok, pid()} | {error, any()}.
  
negative.code: %% @doc Dynamically start a per-topic producer.
  %% Return ok if the producer is already started.
  %% @end
  -spec start_producer(brod:client(), brod:topic(), brod:producer_config()) ->
                          ok | {error, any()}.
  start_producer(Client, TopicName, ProducerConfig) ->
    case get_producer(Client, TopicName, _Partition = 0) of
  