[Bug] Unacked message counter is wrong

[psisoyev]

Search before asking

• I searched in the issues and found nothing similar.

Version

Both 2.11.2 and 3.1.1 images

Minimal reproduce step

1. Start Pulsar standalone docker image
2. Create a Producer publishing to a non-persistent topic
3. Create a Shared Consumer that listens to the topic, consumes messages, ack messages

Code example:

import org.apache.pulsar.client.api.PulsarClient;
import org.apache.pulsar.client.api.Producer;
import org.apache.pulsar.client.api.Schema;

public class PulsarProducer {

    public static void main(String[] args) throws Exception {
        PulsarClient client = PulsarClient.builder()
                .serviceUrl("pulsar://localhost:6650")
                .build();

        Producer<String> producer = client.newProducer(Schema.STRING)
                .topic("non-persistent://public/default/my-topic")
                .create();

        while (true) {
            System.out.println("Sending message...");
            String msg = """
                    { "foo": "bar" }""";
            producer.send(msg);
            Thread.sleep(5000); // Sleep for 5 seconds
        }

        // Never reaches here in this example
        // producer.close();
        // client.close();
    }
}

import org.apache.pulsar.client.api.*;

public class PulsarConsumer {

    public static void main(String[] args) throws Exception {
        PulsarClient client = PulsarClient.builder()
                .serviceUrl("pulsar://localhost:6650")
                .build();

        Consumer<String> consumer = client.newConsumer(Schema.STRING)
                .topic("non-persistent://public/default/my-topic")
                .subscriptionName("my-shared-subscription")
                .subscriptionType(SubscriptionType.Shared)
                .subscribe();

        while (true) {
            // Wait for a message
            Message<String> msg = consumer.receive();
            try {
                System.out.printf("Message received: %s\n", new String(msg.getData()));
                // Acknowledge the message so that it can be deleted by the message broker
                consumer.acknowledge(msg);
            } catch (Exception e) {
                consumer.negativeAcknowledge(msg);
            }
        }

        // Never reaches here in this example
        // consumer.close();
        // client.close();
    }
}

What did you expect to see?

When I run

pulsar-admin topics stats "non-persistent://public/default/my-topic"

I expected to see unackedMessages equal to zero

What did you see instead?

Instead, I'm seeing unackedMessages not equal to zero but it's growing with every processed message

Anything else?

No response

Are you willing to submit a PR?

• I'm willing to submit a PR!