kafka reader drops messages when context with timeout is set.

[xiaofeige]

Describe the bug
my kafka produce 5000 msgs per minutes, when I set a context with 5s timeout, it will drop 1~2 messages in every 5 minutes.
when I increase the timeout to 15s, the dropout msg will decrease. and when I set timout to 60s. no messages will be dropped.

Kafka Version
the latest stable one.

To Reproduce
my kafka produce 5000 msgs per minutes, when I set a context with 5s timeout, it will drop 1~2 messages in every 5 minutes.

Expected behavior
messages shouldn't be drop.

Additional context

I suspect here is the problem.

The message actually is successfully pulled, but the context is timeout when commit message。 but this function returns a empty message and a error with a deadline-exceed.

[gonejack]

Could you show the code you commiting message offsets?

外部是怎么提交offset的呢，看起来在这里设置超时并不会影响offset的正确性。

[achille-roussel]

Hello @xiaofeige, thanks for opening the issue.

I believe the error may be triggered in CommitMessages, which would could cause the behavior you describe since the messages was fetched and committed but the context was cancelled asynchronously.

This is the expected behavior of ReadMessage, if you need to, you can call FetchMessage + CommitMessages in your program and make finer-grain decisions about what to do in the event of a failure (retry, wait, drop the message...).

Unfortunately there's no one-size-fits-all when it comes to reliability trade offs, so the approach we took was to expose lower-level constructs like Fetchmessage and CommitMessages so programs can adopt the behavior that suits their needs.

[xiaofeige]

Could you show the code you commiting message offsets?

外部是怎么提交offset的呢，看起来在这里设置超时并不会影响offset的正确性。

谢谢你的回答，我是使用了consumer group,他这里会自动commit offset

[xiaofeige]

Could you show the code you commiting message offsets?

外部是怎么提交offset的呢，看起来在这里设置超时并不会影响offset的正确性。

这里实际上确实会出现丢消息的情况。看维护者的回答应该是一个做了取舍，这个属于预期内的结果。
丢消息的发生，主要是commit message的时候，context超时，kafka端offset已经commit成功了。但是下面return Message{}, err 意味着返回了错误和一个空值。 这样消息确实就丢了。 看了维护者的建议，准备自己使用 FetchMessage & CommitMessages实现自己的逻辑了。

[xiaofeige]

Hello @xiaofeige, thanks for opening the issue.

I believe the error may be triggered in CommitMessages, which would could cause the behavior you describe since the messages was fetched and committed but the context was cancelled asynchronously.

This is the expected behavior of ReadMessage, if you need to, you can call FetchMessage + CommitMessages in your program and make finer-grain decisions about what to do in the event of a failure (retry, wait, drop the message...).

Unfortunately there's no one-size-fits-all when it comes to reliability trade offs, so the approach we took was to expose lower-level constructs like Fetchmessage and CommitMessages so programs can adopt the behavior that suits their needs.

thanks for your reply. I solved this problem by implementing a new method which uses FetchMessage & CommitMessages. thank a lot.

[achille-roussel]

You're welcome, I'm glad you were able to find a solution 👍