[fix][broker] Fix dispatcher permit accounting race with recycled batch index acks - #26809

Motivation

A Shared or Key_Shared dispatcher keeps totalAvailablePermits, which should always equal the sum of its
consumers' permits. It uses this total to decide how much to read. On every send, the dispatcher and the consumer
each subtract the number of messages delivered. Batch-index acks are on by default. When a partially acknowledged
batch is redelivered, only its unacked messages are delivered, so both subtract messages in batch - acked indexes.
The acked-index count comes from a pooled EntryBatchIndexesAcks object.

The dispatcher read that count after Consumer.sendMessages() returned. By then sendMessages() has handed the
object to the consumer connection's event loop. That task writes the messages and then recycles the object, which
clears every slot. The dispatcher runs on a different thread, so nothing orders the two:

dispatcher thread                                  consumer connection event loop
-----------------                                  ------------------------------
fill batchIndexesAcks (4 of 10 indexes acked)
consumer.sendMessages(...)
  consumer permits -= 10 - 4 = 6   (correct)
  post the write task  ──────────────────────────▶ write the 6 messages
                                                   batchIndexesAcks.recycle()  (clears all slots)
batchIndexesAcks.getTotalAckedIndexCount() → 0
totalAvailablePermits -= 10 - 0    (should be 6)

When the event loop gets there first, the dispatcher subtracts every message in the batch. Its total then falls
below the sum of the consumers' permits, by 4 in this example. The error adds up with each occurrence and is only
reset when all consumers disconnect. Until then, reads are smaller than the consumers can take. The dispatcher is
also reading, without synchronization, an object that another thread is clearing.

The same pattern is in three places:

• PersistentDispatcherMultipleConsumers.trySendMessagesToConsumers
• PersistentDispatcherMultipleConsumers.sendChunkedMessagesToConsumers, used for any read that contains a chunk
• PersistentStickyKeyDispatcherMultipleConsumers.trySendMessagesToConsumers

Consumer-level permits are correct, because Consumer.sendMessages() reads the count before the handoff.

Modifications

• In the three dispatch paths, read batchIndexesAcks.getTotalAckedIndexCount() into a local variable before
  calling sendMessages(). Use it for the permit update and the debug log, so the dispatcher no longer touches the
  object after the handoff. Without a race the result is unchanged, because sendMessages() never modifies the
  object.
• Add SharedDispatcherPermitAccountingTest.testRedeliveredPartiallyAckedBatchDoesNotLosePermits. The test's
  connection stub recycles batchIndexesAcks before sendMessages() returns. That is the losing interleaving, so
  the race reproduces every time. The test dispatches a redelivered batch of 10 messages with 4 acked indexes
  through each of the three paths. It then checks that the dispatcher total equals the sum of the consumers'
  permits. Without the fix, the total is 4 permits short on every path.

Verifying this change

• Make sure that the change passes the CI checks.

This change added tests and can be verified as follows:

• SharedDispatcherPermitAccountingTest.testRedeliveredPartiallyAckedBatchDoesNotLosePermits covers the Shared,
  Shared chunked and Key_Shared paths. It fails without the fix (for example, expected: 194 but was: 190) and
  passes with it.
• The existing dispatcher, Key_Shared, batch-index-ack and Shared chunking suites pass, including
  PersistentDispatcherMultipleConsumersTest, PersistentStickyKeyDispatcherMultipleConsumersTest,
  KeySharedSubscriptionTest, BatchMessageWithBatchIndexLevelTest, BatchMessageIndexAckTest and
  MessageChunkingSharedTest.

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

If the box was checked, please highlight the changes

• Dependencies (add or upgrade a dependency)
• The public API
• The schema
• The default values of configurations
• The threading model
• The binary protocol
• The REST endpoints
• The admin CLI options
• The metrics
• Anything that affects deployment