[fix][broker]Fix repeated messages of shared streaming dispatcher

[poorbarcode]

Fixes #16795

Motivation

In class PersistentDispatcherMultipleConsumers, there is a variable sendInProgress that is set to true before the sendMessagesToConsumers method is executed and to false after the sendMessagesToConsumers method is executed, thus making the readMoreEnties inaccessible during the sendMessagesToConsumers method's execution. see: #16812

In subclass PersistentStreamingDispatcherMultipleConsumers, when reading batch entries, sendMessagesToConsumer is called each entry is read, which causes the variable sendInProgress to be set to true and false multiple times. This results in the method readMoreEnties can be accessed when the variable sendInProgress is set to false when a batch of entries is reading. This makes repeat delivery of the message, the error process is as follows:

step	consumer-1	consumer-2
1	try load messages
2	peek message from the queue messagesToRedeliver	try load messages
3	repay messages	peek message from the queue messagesToRedeliver
4	send message to consumer-1	repay messages
5		send message to consumer-2

Modifications

Set sendInProgress to false only after the last entry in a batch of messages has been delivered.

Documentation

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

Matching PR in forked repository

PR in forked repository:

• [ci][fix][broker]fix Repeated messages of shared streaming dispatcher poorbarcode/pulsar#34

[codelipenghui]

Why we need to remove the consumer name here?

[poorbarcode]

Already fixed

[codecov-commenter]

Codecov Report

Attention: Patch coverage is 0% with 6 lines in your changes missing coverage. Please review.

Project coverage is 41.47%. Comparing base (0866c3a) to head (11f101a).
Report is 2237 commits behind head on master.

Files with missing lines	Patch %	Lines
...ersistentStreamingDispatcherMultipleConsumers.java	0.00%	6 Missing ⚠️

Additional details and impacted files

@@             Coverage Diff              @@
##             master   #18289      +/-   ##
============================================
+ Coverage     38.97%   41.47%   +2.50%     
- Complexity     8311     8864     +553     
============================================
  Files           683      687       +4     
  Lines         67325    67384      +59     
  Branches       7217     7218       +1     
============================================
+ Hits          26239    27949    +1710     
+ Misses        38079    36344    -1735     
- Partials       3007     3091      +84     

Flag	Coverage Δ
unittests	41.47% <0.00%> (+2.50%)	⬆️

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

Files with missing lines	Coverage Δ
...ersistentStreamingDispatcherMultipleConsumers.java	0.00% <0.00%> (ø)

... and 99 files with indirect coverage changes