-
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
[fix][broker] Fix NonDurable Subscription msgBackLog incorrect after topic unload #23305
base: master
Are you sure you want to change the base?
Conversation
pulsar-broker/src/test/java/org/apache/pulsar/client/api/NonDurableSubscriptionTest.java
Outdated
Show resolved
Hide resolved
691386d
to
80c923e
Compare
re-run the failed test |
@@ -1157,7 +1158,8 @@ private CompletableFuture<? extends Subscription> getNonDurableSubscription(Stri | |||
long entryId = msgId.getEntryId(); | |||
// Ensure that the start message id starts from a valid entry. | |||
if (ledgerId >= 0 && entryId >= 0 | |||
&& msgId instanceof BatchMessageIdImpl) { | |||
&& msgId instanceof BatchMessageIdImpl | |||
&& (msgId.getBatchIndex() >= 0 || resetIncludeHead)) { |
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.
Adding resetIncludeHead
should start a PIP. right now, we can only fix by adding msgId.getBatchIndex() >= 0
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.
But if we just add the msgId.getBatchIndex() >= 0
, test of method ReaderBuilder#startMessageIdInclusive
will fail
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.
I will draft a pip to fix this issue later
80c923e
to
54b4feb
Compare
54b4feb
to
e3bfce6
Compare
Fixes #23239
Motivation
Briefly describe the issue:
Consumer consumed all messages and then the msgBacklog is 0, but after topic unload, the msgBacklog turned to be 1 instead of 0 which is not what we expected. The detail is in test NonDurableSubscriptionTest#testNonDurableSubscriptionBackLogAfterTopicUnload.
Root cause:
I analysed the related code, and found this issue was introduced by #4331. In the #4331,
if (((BatchMessageIdImpl) msgId).getBatchIndex() >= 0) {
was deleted directly, which cause none batch message will also execute the stepentryId = msgId.getEntryId() - 1;
. Because of this, the entryId reduced by 1, then the msgBacklog turned to be 1 instead of 0.In the #4331, we added
ReaderBuilder#startMessageIdInclusive
interface to allow create Reader which read containes startMessageId, but it has not been implemented correctly. We should add the field resetIncludeHead in CommandSubscribe to implement it.Modifications
(msgId.getBatchIndex() >= 0 || resetIncludeHead)
, entryId -1 will execute Only when msg is batch or the resetIncludeHead is true.Documentation
doc
doc-required
doc-not-needed
doc-complete
Matching PR in forked repository
PR in forked repository: