[BUG] Shutdown of service leads to loss of message which did not finish processing before shutdown

[pranavendra]

When there is a message for which the mapper function has started processing, and the service using ziggurat is shut down, the mapper function execution for the message stops without running the complete function.

When the service is started again, the message is not reprocessed by the service, leading to loss of the message. This issue was seen on ziggurat v2.12.4

Steps to reproduce

1. Consume a message with a long execution time mapper-fn.
   Add a log at the start of the mapper fn and at the end of the mapper fn.
2. Restart the service immediately.
3. Message which did not fully process due to shutdown, will not be reprocessed by ziggurat.

[mjayprateek]

It is known that this issue occurs while processing messages from RabbitMQ. @pranavendra Have you seen this issue while consuming from Kafka as well?

[mjayprateek]

We're picking this bug up.
From our analysis, the root cause is ziggurat.messaging.consumer/convert-and-ack-message. It acks the message before passing it to the wrapped-mapper-fn i.e. ziggurat.mapper/mapper-func.

This works fine in most of the cases as ziggurat.mapper/mapper-func does its own error handling and pushes the message to retry queue on exception. But it fails when the service stops before the processing is complete.

Solution:
We analysed 2 approaches for fixing this.

1. Acking the message after wrapped-mapper-func completes processing. This is the most straightforward approach.
2. We reject the message if the processing of the message is incomplete. This approach won't work in this particular scenario as if the process stops while mapper-func is running, reject will never be called.

So, we are going ahead with the first approach