[Issue 8260] Support reset cursor to a batch index of the batching message

[Renkai]

Signed-off-by: Renkai gaelookair@gmail.com

(If this PR fixes a github issue, please add Fixes #<xyz>.)

Fixes #8260

Motivation

Make reset cursor command able to reset to a specific index in batch mode.

Modifications

Now reset offset command change index info on broker side

Verifying this change

• Make sure that the change passes the CI checks.

This change added tests and can be verified as follows:

(example:)

• Added integration tests for end-to-end deployment with large payloads (10MB)
• Extended integration test for recovery after broker failure

Does this pull request potentially affect one of the following parts:

If yes was chosen, please highlight the changes

• Dependencies (does it add or upgrade a dependency): (no)
• The public API: (yes / no)
• The schema: (no)
• The default values of configurations: (no)
• The wire protocol: (no)
• The rest endpoints: (no)
• The admin cli options: (no)
• Anything that affects deployment: (no)

Documentation

• Does this pull request introduce a new feature? (no)
• If yes, how is the feature documented? (not applicable / docs / JavaDocs / not documented)
• If a feature is not applicable for documentation, explain why?
• If a feature is not documented yet in this PR, please create a followup issue for adding the documentation

[Renkai]

Still have some problem, unit test written by me can't be passed(due to timeout),
https://github.com/apache/pulsar/pull/8285/files#diff-30ab326ad2d2e0d984b321af3c68c7bb7ab345fac245bbcc6739f58daffc0e68R120

Maybe there is some thing wrong with my test configuration

mvn -pl pulsar-broker -Dtest=SubscriptionSeekTest\#testSeekForBatch test -DredirectTestOutputToFile=false

PTAL @codelipenghui

[jiazhai]

/pulsarbot run-failure-checks

[Renkai]

/pulsarbot run-failure-checks

[codelipenghui]

/pulsarbot run-failure-checks

[codelipenghui]

I have double-checked the implementation of the Batch Index Acknowledgment. Looks we can only pass the ackset to the broker from the client-side, because currently true in the BitSet represent the message does not acked, and false means the batch index are acked. So I think it's the root cause of why the consumer start receive message from the next message of the reset position. You can refer to

pulsar/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PersistentAcknowledgmentsGroupingTracker.java

Lines 188 to 190 in 59e0cfb

	BitSetRecyclable bitSet = BitSetRecyclable.create();
	bitSet.set(0, batchSize);
	bitSet.clear(0, batchIndex + 1);

to check the current implementation.

[Renkai]

/pulsarbot run-failure-checks

[Renkai]

/pulsarbot run-failure-checks

[wolfstudy]

ping @codelipenghui PTAL again thanks.

[Renkai]

/pulsarbot run-failure-checks

[codelipenghui]

cherry-picked to branch-2.6(2.6.2)