fix(filter): Handle MessageRejected correctly (INC-584) #313
Conversation
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
When joining a FilterStep, it can happen that the upstream strategy raises MessageRejected. We need to deal with that somehow, and in this PR I'm just suppressing that exception. Also fix a severe bug where MessageRejected was propagated through submit(), but the MessageRejected was for the wrong message, causing double-submission of messages.
untitaker
changed the title
lynnagara
reviewed
Dec 8, 2023
Comment on lines
92
to
96
| if policy is not None and policy.should_commit(now, self.__uncommitted_offsets): | ||
| # We cannot let MessageRejected propagate here. The caller will | ||
| # think it is for `message` (which has already been successfully | ||
| # submitted), and will double-send it. | ||
| self.__flush_uncommitted_offsets(now, can_backpressure=False) |
There was a problem hiding this comment.
I think this can be removed since you flush on line 78-79 now
There was a problem hiding this comment.
I added this so that the tests continue to pass. But the tests were just overly strict in the behavior we assert. We don't actually need this exact behavior where should_commit is called after every submit
lynnagara
approved these changes
Dec 8, 2023
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
When joining a FilterStep, it can happen that the upstream strategy
raises MessageRejected. We need to deal with that somehow, and in this
PR I'm just suppressing that exception.
Also fix a severe bug where MessageRejected was propagated through
submit(), but the MessageRejected was for the wrong message, causing
double-submission of messages.