Micronaut Kafka Listener commits messages at shutdown that should not be commited with offset strategy SYNC

[JoernSchimmelpfeng]

With OffsetStrategy set to SYNC or ASYNC the KafkaListner should not commit messages when an Exception occurs. However during shutdown all messages get committed regardless if there was an Exception or not.

Task List

• [X ] Steps to reproduce provided
• Stacktrace (if present) provided
• Example that reproduces the problem uploaded to Github
• [ X] Full description of the issue provided (see below)

Steps to Reproduce

1. Have an Micronaut application with a KafkaListener in SYNC mode listening to a Kafka topic.
2. Submit an event to this topic.
3. Start process the event but have an Exception thrown during processing (like a SqlException during a database operation).
4. You now have one message lag on your consumer group as you can validate on Kafka. (as documented)
5. Stop the application
6. You now have a zero message lag on your consumer group as you can validate on Kafka.

Expected Behaviour

The KafkaListener should not commit messages with offset strategy SYNC (or ASYNC) at shutdown time that could not be processed due to an Exception.

Actual Behaviour

All events get committed. Only with offset strategy DISABLED this set is skipped.

[JoernSchimmelpfeng]

Guys this can potentially lead to data loss! Any opinion about this?

[graemerocher]

@burtbeckwith can you please prioritise this issue

[burtbeckwith]

@JoernSchimmelpfeng you checked "example on GitHub" - do you have a test app?

[JoernSchimmelpfeng]

Sorry for my stupidity. I somehow checked the box and could not uncheck it again. Perhaps a browser issue.
However I can provide you with everything that you need, but it would take one or two days. I can also provide a PR if you like since I believe I know the cause. There is a finally block committing everything when leaving the consumer loop.

[graemerocher]

@JoernSchimmelpfeng PRs welcome!

[seancarroll]

I think I've noticed similar behavior. I see JoernSchimmelpfeng added a commit. Is there anything outstanding regarding his proposed fix?

[graemerocher]

@seancarroll can you test the proposed PR and potentially provide a test? it hasn't been merged because it doesn't include a test

[seancarroll]

@graemerocher, given the number of commits behind the PR is where does it make sense to potentially add a test to?

[graemerocher]

You could either rebase the PR or create another PR based off that one and leave the rebase to us

[seancarroll]

Sounds good. I think I have a test that shows for one scenario but it does require an exception to be thrown. In another test that only has a wakeup i'm still seeing behavior in which commits are still taking place. Its unclear to me right now why on a WakeUpException we would ever want to do a commit sync. Or perhaps more specifically why doing a commit in the finally block is required. There also might be something else going in the scenario in which I'm only calling wakeup that is causing commits which I dont think should be occurring. Still attempting to understand the behavior. In particular one example (test) is when using batch and a WakeUpException occurs in the middle of processing.