1) Fix log loss scenario during rebalance, 2) use api version 0 inste… #46
Conversation
…ad of 1, 3) Some minor performance improvement and bug fixes
| Logger.InfoFormat("{2} : Updating fetch offset = {0} with value = {1}", this.fetchedOffset, offset, this.PartitionId); | ||
| this.chunkQueue.Add(new FetchedDataChunk(messages, this, this.fetchedOffset)); | ||
| long newOffset = Interlocked.Exchange(ref this.fetchedOffset, offset); | ||
| Logger.Debug("Updated fetch offset of " + this + " to " + newOffset); |
There was a problem hiding this comment.
Just fixing debug logging to show the right value
There was a problem hiding this comment.
I identified what looked like a different bug here a few months ago: I think line 235 (Interlocked.Exchange...) needs to happen before line 234 (this.chunkQueue.Add...). This puts the this.fetchedOffset value is updated to the new offset before passing it off to the new FetchedDataChunk. Perhaps that fix can be done in a different PR, but I believe it should be done.
Thanks for making the logging change, though. That looks good.
There was a problem hiding this comment.
@mhorstma while editing this file, I had looked at what FetchedDataChunk does with the fetchedOffset. I don't recall all very well, but the order seemed correct. Let us open another PR if we find bugs.
The logging was misleading because the Interchange.Exchange() returns the old value, and not the new offset value. So, I removed the "long newOffset" variable altogether. The debug logs should now be true :)
Major Fix: Log loss scenario which happens when a Kafka connection error happens during re-balance and auto offset reset is enabled to latest
Change requests to use API version 0 instead of 1. This will let the library be usable in 0.9
Minor fixes: Some small logging bugs and performance related fix