Acknowledge manually

[oguzhantortop]

Hi,
From read.me I see only option to manually control ack mechanism is to use reactive flux. I wonder if it is possible to make ack manually with @PulsarConsumer annotation?

[majusko]

This is not possible, and that is the purpose of the library to simplify this subscription and ack concept. There was no need to implement such behaviour with manual consumption yet. Is this something that would be interesting for you?

[oguzhantortop]

Hi, thank you for your response. Yes, it would be. Let me express why I need such control. Think about a scenario in which you consume messages and write them to the database or 3rd party system if you get an error you would like to reprocess the message. Maybe your wrapping code does not give an acknowledge when you get an error I didn't investigate the whole code but anyway it would be a useful addition to give the user additional control...

[majusko]

In that case, any Exception thrown will work because the msg is only acknowledged when the function ends successfully. In case of any exception thrown during the method execution, the message will be acknowledged negatively. If you have correctly configured the retry mechanism, it will also try to process the message again later or send it to the dead letter topic. However, I believe that having manual control can be helpful as well. But your described use case could be easily solved, I believe.

So, in general, the answer is: You can achieve negative ack with any Exception thrown during execution.

[oguzhantortop]

Is this configuration works with the 1.12.0 version ? Because I use a shared subscription and consume a topic of 18 million of records in it. I see that some messages are being delivered again and again. Probably they are not being acknowledged... I tested with a smaller data no problem raised...

pulsar.consumer.default.dead-letter-policy-max-redeliver-count=-1
pulsar.consumer.default.ack-timeout-ms=60000

[majusko]

Please check if you are using the latest library version 1.1.2

TLDR:
If you are receiving the same messages again, it means that the Exception was thrown during the processing or the max ack timeout was exceeded.

---

Throwing an exception or exceeding ack timeout will lead to acknowledging the message negatively. Meaning it will redeliver the message to you again based on pulsar.consumer.default.dead-letter-policy-max-redeliver-count configuration. After the message will be negatively acknowledged X (pulsar.consumer.default.dead-letter-policy-max-redeliver-count) times, the message will be delivered into another topic called the dead letter topic. This mechanism exists in Pulsar to never lose the message.

Since you have the configuration set to -1, it means you don't want to override the default configuration set by Pulsar company, and the value used in their official library is 16.

In case you would like to send messages directly to the dead letter topic, you should set the value to 0.

In case you would like to give the message 1 more shot, you should use the value 1.

The topic's name is created automatically, but you can also configure this for each consumer in the annotation. Also, the number for delivery can be overridden per consumer.

In case you think you found a bug in the library, please try to find a reproduction scenario. I will gladly fix and add this case to automated tests.

[majusko]

There is another case that could cause this - the exception thrown is a type of RuntimeException - as that type of exception is ignored by design.

[oguzhantortop]

There is another case that could cause this - the exception thrown is a type of RuntimeException - as that type of exception is ignored by design.

Don't we give a negative ack in that case?