-
Notifications
You must be signed in to change notification settings - Fork 3.6k
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Fixed race condition while triggering message redelivery after an ack-timeout event #5276
Fixed race condition while triggering message redelivery after an ack-timeout event #5276
Conversation
rerun java8 tests |
rerun java8 tests |
1 similar comment
rerun java8 tests |
run java8 tests |
2 similar comments
run java8 tests |
run java8 tests |
rerun java8 tests |
}); | ||
|
||
for (PositionImpl p : pendingPositions) { | ||
pendingAcks.remove(p.getLedgerId(), p.getEntryId()); |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
pendingPositions
will have same number of positions as pendingAcks
.. so, why can't we just clear pendingAcks
here and anyway, subscription:: redeliverUnacknowledgedMessages
is happening later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
It's to avoid missing any insertion that might be happening into pendingAcks from a different thread. (eg: if the dispatcher thread was actually sending messages at the same time)
rerun java8 tests |
rerun java8 tests |
Motivation
There is a race condition when processing a request for redeliveries after an ack-timeout event. The issue is around the update on the
pendingAcksMap
inConsumer
handler.When we request to re-deliver all messages, we do:
i. go through
pendingAcksMap
and read some of the entriesii. when an entry is read, add the id to
pendingAcksMap
and then dispatchi. go through
pendingAcksMap
to update stats and then clear the mapIf the reading of the entries to be redelivered is very fast (eg. immediately replayed from cache), the clearing might step in after the dispatching thread has re-added the pending entries.
Therefore, at the next iteration, the broker won't re-deliver these messages which will stay in backlog.
Modifications
Ensure that
pendingAcksMap
is correctly updated without a full clear and anyway before any dispatching will take place.