[fix][client] Fix multi-topics consumer could receive old messages after seek

[BewareMyPower]

Motivation

The multi-topics consumer supports seeking all internal consumers when accepting a timestamp as the argument. However, the multi-topics consumer could still read from earliest after seeking to a specific position. There are two reasons:

1. incomingMessages is not cleared before seek(long timestamp)
2. A race condition for a consumer whose start message ID is earliest:
   1. seekAsync was called and then incomingMessages was cleared
   2. Messages from the earliest position were prefetched to internal consumers and then added to incomingMessages
   3. Internal consumers were disconnected from the broker and then broker reset cursors
   4. Messages from the seek position were added to incomingMessages
   5. After seekAsync completed, Consumer#receive will poll the messages fetched before seek completed

Modifications

Add a duringSeek flag that is set with true when the seek operation starts. Then do not handle messages via messageReceive and stop receiveMessageFromConsumer if duringSeek is true. After the seek option finishes, set duringSeek back to false and restart receiveMessageFromConsumer for all consumers again.

Add testSeekToNewerPosition to verify receive, receiveAsync and message listener all work for seeking by timestamp to a newer position on a multi-topics consumer.

Documentation

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

Matching PR in forked repository

PR in forked repository:

[BewareMyPower]

This issue should also exist for other seekAsync overrides, I will refactor the seek logic soon.

[RobertIndie]

Good work! Left one comment.

[BewareMyPower]

@codelipenghui I have addressed your comments, PTAL again.

[codecov-commenter]

Codecov Report

Attention: 5 lines in your changes are missing coverage. Please review.

Comparison is base (17bb322) 73.59% compared to head (1ee4441) 73.67%.
Report is 25 commits behind head on master.

Additional details and impacted files

@@             Coverage Diff              @@
##             master   #21945      +/-   ##
============================================
+ Coverage     73.59%   73.67%   +0.07%     
- Complexity    32417    32489      +72     
============================================
  Files          1861     1863       +2     
  Lines        138678   138784     +106     
  Branches      15188    15207      +19     
============================================
+ Hits         102060   102245     +185     
+ Misses        28715    28651      -64     
+ Partials       7903     7888      -15     

Flag	Coverage Δ
inttests	24.08% <12.50%> (-0.05%)	⬇️
systests	23.65% <12.50%> (-0.03%)	⬇️
unittests	72.94% <84.37%> (+0.07%)	⬆️

Flags with carried forward coverage won't be shown. Click here to find out more.

Files	Coverage Δ
...n/java/org/apache/pulsar/broker/PulsarService.java	83.06% <100.00%> (+0.74%)	⬆️
...ache/pulsar/client/admin/internal/BrokersImpl.java	87.50% <100.00%> (ø)
...g/apache/pulsar/broker/admin/impl/BrokersBase.java	70.86% <50.00%> (ø)
...he/pulsar/client/impl/MultiTopicsConsumerImpl.java	78.41% <84.61%> (+0.31%)	⬆️

... and 113 files with indirect coverage changes