Github user srdo commented on the issue:
    @hmcl I agree that if there really is a double acking problem somewhere 
else, it should be fixed there.
    In your scenario say the spout commits 1...5 to Kafka and 4 is later acked. 
ackedMsgs now contains 4, and committedOffset is 5. With the previous code, 
that would permanently break findNextCommitOffset. 
    findNextCommitOffset would start at 5, and the first message in ackedMsgs 
list is 4. The code breaks out of the loop over ackedMsgs, since the else block 
in L496 is hit, causing the function to return null. When null is returned to 
commitOffsetsForAckedTuples, that means that it will skip that topic partition 
when committing, which means that ackedMsgs doesn't get cleared. That means 
that on the next call to findNextCommitOffset the same thing happens again. The 
end result is that the spout never commits anything for that topic partition 
    I agree that we should never be in the case where 4 is acked after 
committedOffset has become larger than 4, which is why I think the else block 
in L496 should probably log at a higher level than debug.

If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at or file a JIRA ticket
with INFRA.

Reply via email to