[fix][broker] Fix NonDurable Subscription msgBackLog incorrect after topic unload

[summeriiii]

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 step entryId = 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

• CommandSubscribe add the field resetIncludeHead, and when use the ReaderBuilder#startMessageIdInclusive, this param is true while other is false.
• Persist#getNonDurableSubscription method add the judge condition (msgId.getBatchIndex() >= 0 || resetIncludeHead), entryId -1 will execute Only when msg is batch or the resetIncludeHead is true.

               if (ledgerId >= 0 && entryId >= 0
                        && msgId instanceof BatchMessageIdImpl
                        && (msgId.getBatchIndex() >= 0 || resetIncludeHead)) {
                    // When the start message is relative to a batch, we need to take one step back on the previous
                    // message,
                    // because the "batch" might not have been consumed in its entirety.
                    // The client will then be able to discard the first messages if needed.
                    entryId = msgId.getEntryId() - 1;
                }

Documentation

• doc
• doc-required
• doc-not-needed
• doc-complete

Matching PR in forked repository

PR in forked repository:

[summeriiii]

@lovelle @lhotari please take a look :)

[Technoboy-]

re-run the failed test

[Technoboy-]

Adding resetIncludeHead should start a PIP. right now, we can only fix by adding msgId.getBatchIndex() >= 0

[summeriiii]

But if we just add the msgId.getBatchIndex() >= 0, test of method ReaderBuilder#startMessageIdInclusive will fail

[summeriiii]

I will draft a pip to fix this issue later