title: Apache Pulsar Reader can still get messages which should be deleted
body: <p>I am a newbie to Apache Pulsar (also to MQ system). Now, I have a question about Pulsar Reader.</p>
  
  <p><strong>Question description:</strong><br>
  I launch a Pulsar instance, then start a consumer, listen on a topic A.
  Then I start a producer, send 100 messages to topic A, and the consumer consumed the 100 messages, the value of Backlog in consumer's subscription is 0.There is only one subscription on the topic, and it's exclusive.</p>
  
  <p>After that, I start a Reader, set Reader's topic is A, Reader can get messges from topic A.</p>
  
  <p>I found this on Pulsar docs:
  <a href="https://pulsar.apache.org/docs/en/cookbooks-retention-expiry/" rel="nofollow noreferrer">https://pulsar.apache.org/docs/en/cookbooks-retention-expiry/</a></p>
  
  <blockquote>
    <p>Pulsar brokers are responsible for handling messages that pass through Pulsar, including persistent storage of messages. By default, brokers:</p>
    
    <p>immediately delete all messages that have been acknowledged on every subscription,<br>
    and persistently store all unacknowledged messages in a backlog.</p>
  </blockquote>
  
  <p>The 100 messages should alreadly be deleted.
  So why did Pulsar Reader can still get messages from topic A?</p>
  
  <p><strong>my code:</strong>  </p>
  
  <p>consumer:</p>
  
  <pre><code>private static void consume() {
          try {
              PulsarClient pulsarClient = PulsarClient.builder().serviceUrl("pulsar://127.0.0.1:6650").build();
              Consumer&lt;String&gt; consumer = pulsarClient.newConsumer(Schema.STRING)
                      .topic("A")
                      .subscriptionName("first-subscription")
                      .subscribe();
  
              for (int i = 0; true; ++i) {
                  try {
                      Message&lt;String&gt; msg = consumer.receive();
                      String m = msg.getValue();
                      System.out.println("\t m:" + m);
                      consumer.acknowledge(msg);
                      Thread.sleep(500);
                  } catch (Exception e) {
                      LOGGER.error("", e);
                  }
              }
          } catch (Exception e) {
              LOGGER.error("", e);
          }
      }
  </code></pre>
  
  <p>producer:</p>
  
  <pre><code>private static void produce() {
          try {
              PulsarClient pulsarClient = PulsarClient.builder().serviceUrl("pulsar://127.0.0.1:6650").build();
              Producer&lt;String&gt; producer = pulsarClient.newProducer(Schema.STRING).topic("A").create();
              for (int i = 0; i &lt; 100; ++i) {
                  producer.send("producer-simple-partitioned-" + i);
              }
          } catch (Exception e) {
              LOGGER.error("", e);
          }
      }
  </code></pre>
  
  <p>Reader:</p>
  
  <pre><code>private static void readerRead() {
          try {
              PulsarClient pulsarClient = PulsarClient.builder().serviceUrl("pulsar://127.0.0.1:6650").build();
              Reader&lt;byte[]&gt; reader = pulsarClient.newReader()
                      .topic("A")
                      .startMessageId(MessageId.earliest)
                      .create();
              while (true) {
                  Message message = reader.readNext();
                  System.out.println(new String(message.getData()));
              }
          } catch (PulsarClientException e) {
              LOGGER.error("", e);
          }
      }
  </code></pre>