Fix auto-commit bug and make a feature of auto-commit #1687
Conversation
| time.Sleep(time.Second) | ||
| } | ||
| }() | ||
|
|
There was a problem hiding this comment.
What are you doing?! You mark all fetched messages consumed every second? And not respect to the closing event (ctx.Done() or parent.closed), which cause goroutine leak.
There was a problem hiding this comment.
Thanks a lot for you review.
I want to implement the auto-commit feature. Yes, closing event should be respected.
There was a problem hiding this comment.
IIRC, the auto-commit feature already exists.
There was a problem hiding this comment.
See this thread, #1570
The auto-commit feature won't work, and i have add analysis in the desc of this fix.
|
@lqiz thanks for putting this PR together. I think that the functionality you were looking for has now been covered as part of the merged PR #1699 which correctly disables the auto-commit loop if you choose to disable auto-commits, and provides a Feel free to come back if you think there is still some capability missing here though. |
I believe that the sarama-cluster hasn’t designed the auto-commit feature. What we have to do is updating the offset via MarkOffset/MarkMessage after consuming the message.
Here is its reason:
Offset manager is completely independent of the consumer. flushToBroker in main-loop has only flush offset to the broker but it doesn’t update the offset.
It has no relation between the offset updating and the consumer in sarama. Currently, the offset updating depends on the Markoffset after consuming the message.
If auto-commit want to be supported in sarama, it should update the offset when fetching messages. After offset updating, the main-loop will handle commit things left.
So this fix is made via updating the offset.