[fix][broker] Fix replicator stall when a skipped pending read completes after a cursor rewind

[lhotari]

Motivation

PersistentReplicator.readEntriesComplete() has a "skip" branch that runs when the replicator has terminated or when the read result must be discarded because the cursor was rewound (InFlightTask.isSkipReadResultDueToCursorRewind()). A cursor rewind happens, for example, when a new schema is detected and the replicator pauses to fetch it.

When a cursor rewind cancels in-flight reads, cancelPendingReadTasks() only resets the pending reading task to a completed state when cursor.cancelPendingReadRequest() returns true. If the read has already been dispatched to BookKeeper it cannot be cancelled, so the request returns false and the pending task is left with entries == null while still flagged with skipReadResultDueToCursorRewind == true.

That dispatched read still delivers a readEntriesComplete() callback. The skip branch then calls InFlightTask.incCompletedEntries() for each returned entry, but because the task's entries field is still null, incCompletedEntries() cannot increment completedEntries and instead logs Unexpected calling of increase completed entries. The task therefore never becomes isDone(): it permanently occupies the single read slot (getPermitsIfNoPendingRead() returns 0 and hasPendingRead() returns true), so readMoreEntries() can never schedule another read and replication for that direction stalls.

This was observed as a stall at a schema boundary — e.g. flaky failures of ReplicatorTest.testReplicationWithSchema, where messages produced with a second schema were never replicated and the consumers on the remote clusters timed out. The race requires a cursor read to already be dispatched to BookKeeper (so it can no longer be cancelled) at the moment the schema-fetch rewind runs, which is why it reproduces under the slower I/O timing of CI but is hard to reproduce on a fast local machine.

Modifications

• In readEntriesComplete()'s skip branch, record the read result on the in-flight task (setEntries(entries)) before completing it, so the task becomes isDone() and releases its read permit, allowing the read loop to resume.
• Add PersistentReplicatorInflightTaskTest.testSkippedPendingReadDoesNotStallReplication, a deterministic regression test that reproduces the stall (a pending read flagged skip whose dispatched read then completes) and asserts the task is reconciled and read permits are restored.

Verifying this change

This change added tests and can be verified as follows:

• ./gradlew :pulsar-broker:test --tests "org.apache.pulsar.broker.service.persistent.PersistentReplicatorInflightTaskTest.testSkippedPendingReadDoesNotStallReplication" — the new test fails on master (the in-flight task is never completed, so read permits are never restored) and passes with this change.
• ./gradlew :pulsar-broker:test --tests "org.apache.pulsar.broker.service.persistent.PersistentReplicatorInflightTaskTest" and --tests "org.apache.pulsar.broker.service.ReplicatorTest.testReplicationWithSchema" pass.

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

[lhotari]

Closing as a duplicate of #26106, which fixes the same bug (a cursor-rewind-skipped in-flight read leaving the InFlightTask stuck in readEntriesComplete()'s skip branch) more completely.

This PR only completed the task to release its read permit. #26106 additionally:

• rewinds the cursor again, because the discarded read re-advances the ManagedCursor read position past the still-unacked messages after doRewindCursor() ran (otherwise the messages are stranded and the backlog never drains), and
• resumes reads on the broker executor (avoiding inline-cache-hit readEntriesComplete → readMoreEntries recursion / StackOverflowError),

both under the inFlightTasks lock. It also adds an end-to-end recovery test. Superseded by #26106.