[fix][broker] Trigger max read position callback for messages published during transaction buffer recovery

[lhotari]

Motivation

ReplicatedSubscriptionWithTransactionTest.testReplicatedSubscribeAndSwitchToStandbyCluster fails intermittently in CI with:

org.awaitility.core.ConditionTimeoutException: Assertion condition One snapshot should be finished expected [true] but found [false] within 10 seconds.

Example failure: https://github.com/apache/pulsar/actions/runs/29992607875/job/89160931729 (a dependabot PR unrelated to this code path).

The captured test logs show that both clusters' replicators were connected and all test messages were replicated within ~1 second of the test starting, followed by total silence: no replicated-subscription snapshot round trip was ever attempted during the remaining ~9 seconds of the Awaitility window.

Root cause is in TopicTransactionBuffer, not in the test:

• While the transaction buffer is recovering (Initializing state), syncMaxReadPositionForNormalPublish() ignores normal publishes: maxReadPosition does not move and the maxReadPositionMovedForward callback is not invoked.
• When recovery completes, recoverComplete() / noNeedToRecover() catch maxReadPosition up to the managed ledger's last confirmed entry by direct field assignment, without invoking the maxReadPositionMovedForward callback.

As a result, messages published while the transaction buffer is recovering never update PersistentTopic.lastMaxReadPositionMovedForwardTimestamp. ReplicatedSubscriptionsController.startNewSnapshot() uses that timestamp to decide whether the topic has new data; with the timestamp stuck at 0, every snapshot tick is skipped with "There is no new data in topic. Skipping snapshot creation." until some new message is published after recovery.

In the flaky test, the whole publish burst races the transaction buffer recovery of the freshly created topic. When all messages land before recovery completes, no subscription snapshot can ever start (the test publishes nothing more while polling the assertion), so the test times out regardless of the Awaitility timeout. In production the condition self-heals on the next publish, which is why it mainly surfaces as a test flake, but until new traffic arrives a replicated-subscription snapshot is silently delayed the same way.

Modifications

• TopicTransactionBuffer: extract a updateMaxReadPositionAfterRecovery() helper used by both recoverComplete() and noNeedToRecover(). It advances maxReadPosition to the last confirmed entry and invokes the maxReadPositionMovedForward callback if the position actually moved forward while recovery was in progress. When nothing was published during recovery, the position does not move and the callback is not invoked, preserving the existing "no snapshot without traffic" behavior (covered by ReplicatedSubscriptionTest.testReplicationSnapshotStopWhenNoTraffic).
• Add TopicTransactionBufferRecoveryTest, a deterministic regression test that gates transaction buffer recovery on a controllable future (mocked AbortedTxnProcessor.recoverFromSnapshot()), publishes messages while recovery is pending, and asserts that completing the recovery (both with and without a snapshot to replay) moves lastMaxReadPositionMovedForwardTimestamp forward. It also asserts the negative case: recovery of an idle topic must not move the timestamp.

Verifying this change

• Make sure that the change passes the CI checks.

This change added tests and can be verified as follows:

• TopicTransactionBufferRecoveryTest.testMaxReadPositionMovedForwardForMessagesPublishedDuringRecovery deterministically reproduces the bug: it fails without the production change and passes with it.
• TopicTransactionBufferRecoveryTest.testMaxReadPositionNotMovedForwardWhenNothingPublishedDuringRecovery guards against spuriously triggering snapshots for idle topics.
• The previously flaky ReplicatedSubscriptionWithTransactionTest.testReplicatedSubscribeAndSwitchToStandbyCluster passed 10/10 locally with a temporary invocationCount = 10 (annotation not included in this PR).
• Full ReplicatedSubscriptionTest and ReplicatedSubscriptionWithTransactionTest classes pass locally.

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