Is it possible to manually acknowledge messages via msgId?

[crossoverJie]

I didn't find the relevant API here:
https://pulsar.apache.org/docs/2.10.x/admin-api-topics/

So I tried it with the following code:

        PulsarAdmin admin = PulsarAdmin.builder().serviceHttpUrl("http://127.0.0.1:8080")
                .build();
        String topic ="persistent://public/default/topic-test-partition-1";
        Message<byte[]> id = admin.topics().getMessageById(topic, 407, 5);

        PulsarClient client = PulsarClient.builder()
                .serviceUrl("pulsar://127.0.0.1:6650")
                .build();
        Consumer consumer = client.newConsumer()
                .topic("persistent://public/default/topic-test-partition-1")
                .subscriptionName("test")
                .subscriptionType(SubscriptionType.Shared)
                .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest)
                .ackTimeout(0, TimeUnit.SECONDS)
                .subscribe();


        consumer.acknowledge(id);

it did not achieve the desired effect, and I can still query this message through this API.

admin.topics().getInternalStats(topic);

{"individuallyDeletedMessages":"[(407:4..407:5]]"}

Could someone please help to explain this to me?
Thanks.

@labuladong @RobertIndie @Jason918

[labuladong]

Hi @crossoverJie,

it did not achieve the desired effect
{"individuallyDeletedMessages":"[(407:4..407:5]]"}

But I think this is the expected behavior: you acknowledge (mark delete) the message 407:5 so individuallyDeletedMessages records this message as deleted.

Can you describe what's your desired effect?

By the way, you can just use MessageIdImpl to specify a message id:

consumer.acknowledge(new MessageIdImpl(legerID, entryID, partitionIndex));

[michaeljmarshall]

I agree with @labuladong that the behavior described is the expected behavior. The message is able to be retrieved because it has not yet been deleted. However, the consumer on that subscription will not receive that message again. Please confirm that is the case.

Also, note that messages can only be deleted up to to the mark delete position. Messages listed as individuallyDeletedMessages are not eligible for deletion.

Additionally, note that it is recommended to build MessageIdImpl manually. This recent thread on the Apache Pulsar dev mailing list has a long discussion about proper abstraction and what applications are supposed to rely on: https://lists.apache.org/thread/rdkqnkohbmkjjs61hvoqplhhngr0b0sd.

[crossoverJie]

Thanks for the answer,

Before this I thought that individuallyDeletedMessages stored unacknowledged messages.

---

Here is my testing process:

"individuallyDeletedMessages": "[(414:36..414:39]]",
consumer.acknowledge(new MessageIdImpl(414, 35, 1));
consumer.acknowledge(new MessageIdImpl(414, 36, 1));

Query individuallyDeletedMessages again:
"individuallyDeletedMessages": "[(414:36..414:39]]",
The results did not change.

When I restart the consumer:

consumer Message received: 414:35:1:0
consumer Message received: 414:36:1:0

The message that was just manually confirmed is still available, At the same time:
"individuallyDeletedMessages": "[]"

My expected results should be:
After I manually acknowledge the message, the restart consumer does not get the message again and at the same time, individuallyDeletedMessages should change.

Is my understanding correct?

thx.

[crossoverJie]

The background to my doing this is as follows:

I want to know if the backlog of message topics is caused by message holes.

If so, I would like to manually acknowledge these holes in our admin console for the purpose of clearing them.

Have the broker automatically move the MarkDeletedPosition and clear the backlog of messages.

Any other suggestions to make this work?
thx.

[crossoverJie]

Query individuallyDeletedMessages again:
"individuallyDeletedMessages": "[(414:36..414:39]]",
The results did not change.

I'm sorry, that was my mistake.

I was testing in JUnit and the client exited before the background thread finished executing, resulting in the message not being committed.

Now the main function works fine instead.

The background to my doing this is as follows:

I want to know if the backlog of message topics is caused by message holes.

If so, I would like to manually acknowledge these holes in our admin console for the purpose of clearing them.

Have the broker automatically move the MarkDeletedPosition and clear the backlog of messages.

Any other suggestions to make this work? thx.

By the way, do Any suggestions in this background? Thanks

[labuladong]

For stats of ack holes, this PR #18263 may help, please take a look. And to handle ack holes, maybe accumulative ack can help.