[fix][client] Prevent epoch race in MultiTopicsConsumer batch receive

[ChimdumebiNebolisa]

Fixes #25204

Motivation

MultiTopicsConsumer processes a received batch by validating the consumer epoch per message. If redeliverUnacknowledgedMessages() runs concurrently, it can increment consumerEpoch while the batch loop is still iterating, which can produce a mixed outcome where part of the batch is accepted and the rest is filtered. This matches the behavior reported in #25204.

Modifications

• Serialized batch processing with redeliver by holding incomingQueueLock across the entire batch loop and the incomingMessages.size() read, so consumerEpoch cannot change mid-batch.
• Updated MultiTopicsConsumerEpochRaceTest to assert the post-fix invariant:
  • redeliver is blocked while the batch holds the lock
  • the batch observes a stable epoch and accepts both messages (acceptedByEpochCount == 2)
  • redeliver completes shortly after the batch releases the lock

Verifying this change

• This change added tests and can be verified as follows:
  • Run the regression test (it is in flaky, so it must not be excluded):
    • mvn --% -pl pulsar-client "-Dtest=org.apache.pulsar.client.impl.MultiTopicsConsumerEpochRaceTest" "-DexcludedGroups=quarantine" test
  • Run module tests:
    • mvn -pl pulsar-client test

Does this pull request potentially affect one of the following parts

• The threading model

Documentation

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

Matching PR in forked repository

PR in forked repository: ChimdumebiNebolisa#2

[github-actions]

@ChimdumebiNebolisa Please add the following content to your PR description and select a checkbox:

- [ ] `doc` <!-- Your PR contains doc changes -->
- [ ] `doc-required` <!-- Your PR changes impact docs and you will update later -->
- [ ] `doc-not-needed` <!-- Your PR changes do not impact docs -->
- [ ] `doc-complete` <!-- Docs have been already added -->

[BewareMyPower]

You should not commit the investigation document to the code base. You can post it to Gist and share the link in the PR description

[ChimdumebiNebolisa]

Thanks for the review.

I’ve updated the test to address the feedback:

• Removed reflection for consumers and access it directly (it’s protected).
• Removed the flaky grouping and eliminated Thread.sleep by coordinating with latches.
• Replaced assertTrue(isEmpty()) with assertEquals(size, 0) and removed the redundant ConnectionHandler import.

The remaining reflection is only for private members with no stable test seam (startReceivingMessages, resumeReceivingFromPausedConsumersIfNeeded, and MessageImpl.consumerEpoch). The test passes deterministically.