[Bug Report]: KafkaBus.StopAsync() blocks and does not return if there are messages still pending to be consumed about kafkaflow HOT 3 CLOSED

Hi @joelfoliveira,
Yes, I am. I’ve been talking with @filipeesch and we’ve found out a corner case where the message is processed faster than it is inserted in the offset manager, making it throw an exception when trying to complete it, and thus never completing it (nor storing the offsets).
We have a fix but are still deciding the best approach to implement it.

from kafkaflow.

This behavior is expected, because KafkaFlow will try to complete the message processing before stopping in order to prevent duplication of message processing and to commit the offset accordingly (graceful shutdown).

In cases, where the message requires a long-running task to process you can use the WorkerStopped CancellationToken available through the IMessageContext (messageContext.ConsumerContext.WorkerStopped) to listen for the cancellation and to interrupt the message processing. By default, this token is canceled 30 seconds after the worker has been issued to stop processing.
The time it takes for the cancellation can be changed using the Consumer Configuration method .WithWorkerStopTimeout(...)

from kafkaflow.

Hi @massada ,

If you're still experiencing this issue or have any additional information to share, please feel free to let us know.

However, if we don't receive any updates or feedback from you within the next 5 business days, we may need to consider closing this issue. Please understand that this is not a final decision, and you can always reopen the issue or create a new one in the future.

We appreciate your contribution to our project and look forward to hearing from you soon. If you have any questions or need further assistance, don't hesitate to reach out.

from kafkaflow.