Support for long running message consumer

[shubham-Shole4ever]

Is your feature request related to a problem? Please describe.
The ackTimeout is set at the consumer level and is valid for all the messages that consumer handles. We have a case where the consumption of a message takes an unpredictable amount of time, ranging from 10 mins to couple hours. We also don't want to set the ackTimeout for the messages to be max possible (which could be half a day or more).
Can we have a feature where the consumer can send back a signal to the broker, acknowledging that its not failed but currently working on the received message, and the broker extends the ackTimeout for that message.

Describe the solution you'd like
A functionality which allows the consumer to notify the broker that it is working on the received message. The broker, on receiving this signal can extend the ackTimeout for that particular message (probably refreshing the ackTimeout)

Describe alternatives you've considered
Currently, there is no way to modify the ackTimeout for a particular message. The ackTimeout is set at the consumer level and cannot be modified for any message.

[codelipenghui]

Please take a look at this document which may help you. http://pulsar.apache.org/docs/en/concepts-messaging/#negative-acknowledgement

[shubham-Shole4ever]

@codelipenghui I had a look at the negative-acknowledgement. This will still not work if my ackTimeout is set to 10 mins and the message I am consuming is taking 30 mins (for e.g.). The broker will resurface the message after 10 mins, even though one of the consumer is still working on it.
I want to avoid this scenario. My proposal is to have something like a "working(messageId)" functionality on the consumer, which notifies the broker not to timeout (and resurface) the message, but rather extend/refresh the ackTimeout set for the concerned messageId.

[codelipenghui]

@shubham-Shole4ever
You can disable ack timeout, just use ack/negative ack. It means explicitly telling the broker that the process failed and then the broker redeliver this message, if message is in progress, no need to ack/negative ack.

[shubham-Shole4ever]

@codelipenghui
But if my application crashes while it is processing the message, it'll never be able to ack/negative ack that message ever. This'll result in that message never being retried. This is the exact scenario why I cannot ditch the ackTimeout as well.

[merlimat]

@shubham-Shole4ever When a consumer crashes, or the TCP connection is broken, the messages that were delivered to this consumer and not acked, will be replayed to another available consumer (in case of shared subscriptions) or next time the consumer reconnects.

You don't need ack timeout for that.

[sijie]

@shubham-Shole4ever does Matteo's comment make sense to you?

[shubham-Shole4ever]

@sijie I can do with the workaround suggested by @codelipenghui and @merlimat for the time being. However, as mentioned, the solution will not work in case I also have a need of ackTimeout.
Would request the community to propose a feature to handle such cases for future.

Thanks @codelipenghui and @merlimat for all the help. :)

[harissecic]

Stumbled upon this looking for another answer. However in case it helps anyone I'll leave a comment. For such cases I guess it's possible to combine DLQ with ackTimeout. Default value I use is 3. Although I don't use auto-ack I guess it will still work the same way. If message times-out 3 times (in this case) it will automatically go to Dead Letter Queue. This will prevent it to loop endlessly between services. My example is that I'm building up a module for a framework. Now in such case I don't do ackTimeout but let users set it if they want to. However, I do by default set 3 retries before DLQ. Reason was personal experience where it endlessly looped my test message to the shared consumers and I got error logs all the time and couldn't figure it out. Then I realised well message is simply getting negativeAck from each consumer and then redelivered all the time but funny thing is it was malformed JSON message so consumers were doomed to crash (validations falied for previously nullable thing in kotlin that I moved to non-null). When I set up DLQ to 3 I had some messages fail and then get re-read due to timeout for ack. But combining DLQ, ackTimeout, and shared consumers I think you can set timeout pretty low if processing data takes less time and you do manual ACK as soon as it's done.