[fix][broker] Fix replication stall when a cursor rewind skips an in-flight read

[lhotari]

Motivation

When a publish to the remote cluster fails, PersistentReplicator rewinds the cursor
(beforeTerminateOrCursorRewinding(Failed_Publishing) + doRewindCursor) so the unacked messages
are re-read. If a cursor read was already dispatched at that moment, cursor.cancelPendingReadRequest()
returns false (there is no waiting read op to cancel — the common case when there is a backlog), so
cancelPendingReadTasks() only flags the in-flight task and does not complete it.

When that dispatched read later completes, readEntriesComplete() discarded the result in the skip
branch but left the task with entries == null. This caused a two-fold stall:

1. The task stayed entries == null forever, so hasPendingRead() stayed true and
   getPermitsIfNoPendingRead() returned 0 — no further reads were ever issued.
2. ManagedCursor advances the read position when a read completes
   (OpReadEntry → ManagedCursorImpl.setReadPosition), so the discarded read moved the read position
   past the still-unacked messages, overwriting the rewind from doRewindCursor(). Even once reads
   resumed, those messages were stranded behind the read position and never re-read.

The geo-replication backlog therefore never drained. This is independent of the replicator dispatch
rate-limiter path; it exists on master regardless of #26005 (it was found while reviewing that PR).

The same defect is reachable from any cursor rewind that races an already-dispatched read, not only a
failed publish — in particular the rewind performed when a new schema is detected
(beforeTerminateOrCursorRewinding(Fetching_Schema) + doRewindCursor(true) in
GeoPersistentReplicator.replicateEntries). It has been observed in CI as flaky failures of
ReplicatorTest.testReplicationWithSchema, where replication stalled at the PersonOne →
PersonThree schema boundary and the remote-cluster consumers timed out. A concrete recent occurrence
is this run (on an
unrelated PR whose branch is based on master without this fix): Broker Group 5 failed in
testReplicationWithSchema with AssertionError: expected [true] but found [false] at
ReplicatorTest.java:414 — assertTrue(msg1 != null && msg2 != null && msg3 != null) — because a
remote consumer's receive(10, SECONDS) returned null for a message past the schema boundary. That
race only reproduces under the slower I/O timing of CI (a read must already be dispatched to bookies
when the rewind runs), which is why it is hard to reproduce on a fast local machine; the added unit/e2e
tests reproduce it deterministically. (No separate flaky-test tracking issue was filed for that
symptom.)

Modifications

In the readEntriesComplete() skip branch, when the read result is discarded due to a cursor rewind
and the replicator is not terminating:

• Rewind the cursor again so the messages stranded by the completing read are re-read, and complete the
  task with an empty result so it releases its permit and stops counting as a pending read. Both are
  done under synchronized(inFlightTasks) — the same lock doRewindCursor() uses — so they are atomic
  with respect to readMoreEntries().
• Resume reading on the broker executor instead of calling readMoreEntries() directly, to avoid deep
  readEntriesComplete -> readMoreEntries recursion when a cursor read completes inline
  (StackOverflowError), the same hazard PersistentDispatcherMultipleConsumers.readMoreEntriesAsync()
  guards against.

This also removes the per-entry incCompletedEntries() call the skip branch previously made. That call
could never complete the task: incCompletedEntries() only advances completedEntries when the
task's own entries field is non-empty, but at this point in readEntriesComplete() the read
result exists only as the method parameter — the task's entries field is assigned later, in the
normal (non-skip) path that this branch returns before reaching, so it is still null here. The call
therefore always fell into the else branch, logging Unexpected calling of increase completed entries once per entry (these are the errors seen in the stalled-replication logs) without
incrementing anything or making isDone() return true. In other words it looked like it completed
the task but did nothing, which is the core reason the task leaked. Completing the task with
setEntries(Collections.emptyList()) (above) is what actually marks it done — isDone() returns
true for an empty result — so the permit is released; the entry buffers are still freed via
entry.release().

Note on semantics: the recovery rewind resets the read position to the mark-delete position, so only
unacked messages are re-read — entries already acknowledged (replicated) are skipped on the
re-read. Recovery is therefore at-least-once, with duplicates bounded to the messages that were in
flight when the rewind happened (the prior behavior delivered nothing at all on this path).

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.testCursorRewindSkippedReadCompletesInFlightTask"
  — a deterministic unit test that a skip-flagged in-flight read completes the task and frees the read slot.
• ./gradlew :pulsar-broker:test --tests "org.apache.pulsar.broker.service.persistent.PersistentReplicatorInflightTaskTest.testReplicationRecoversAfterPublishFailureRewindWithInflightRead"
  — an end-to-end test over two clusters that holds a real cursor read in flight, rewinds the cursor the
  same way a failed publish does (Failed_Publishing + doRewindCursor(false)), then asserts the
  backlog drains and every produced message is delivered to the remote cluster exactly once (no loss, no
  duplicates).
• ./gradlew :pulsar-broker:test --tests "org.apache.pulsar.broker.service.persistent.PersistentReplicatorInflightTaskTest.testReplicationRecoversAfterSchemaFetchRewindWithInflightRead"
  — an end-to-end test that holds a real cursor read in flight and rewinds the cursor through the
  schema-fetch path: the same two rewind calls (beforeTerminateOrCursorRewinding(Fetching_Schema)
  • doRewindCursor(true)) that GeoPersistentReplicator.replicateEntries issues at a schema boundary,
    then asserts the backlog drains and every produced message is delivered to the remote cluster. This is
    the deterministic counterpart to the flaky ReplicatorTest.testReplicationWithSchema. The rewind is
    driven directly (rather than by crossing a real schema boundary) because the bug needs a read to be in
    flight at the exact moment the rewind runs, which on the natural path depends on a read being pipelined
    — dispatched by an earlier message's sendComplete — concurrently with replicateEntries reaching the
    schema boundary, an inherently racy interleaving. The test therefore guards the readEntriesComplete
    skip-branch recovery for the Fetching_Schema rewind, not the schema-detection wiring itself.

All three tests fail on master (the e2e tests time out with the backlog stuck at the full message
count) and pass with the fix; the e2e tests were run repeatedly (invocationCount 5–10) with no
flakiness.

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

• 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

[void-ptr974]

LGTM. I checked the skipped in-flight read path after cursor rewind: the task is now completed so permits are released, the cursor is rewound again to undo the read-position advance from the discarded read, and reads resume asynchronously to avoid recursion. The added unit and e2e replication tests cover both the leaked pending task and backlog recovery. Nice, precise fix for a subtle race.

[lhotari]

Recent CI failure where this production issue showed up: https://github.com/apache/pulsar/actions/runs/28466180778/attempts/1?pr=26122

That run is on an unrelated PR (#26122, a different flaky-test fix) whose branch is based on master without this fix. The CI - Unit - Brokers - Broker Group 5 job failed in ReplicatorTest.testReplicationWithSchema:

java.lang.AssertionError: expected [true] but found [false]
    at org.apache.pulsar.broker.service.ReplicatorTest.testReplicationWithSchema(ReplicatorTest.java:414)

Line 414 is assertTrue(msg1 != null && msg2 != null && msg3 != null) — a remote-cluster consumer timed out (receive(10, SECONDS) returned null) on a message after the schema boundary. The test sends 500 messages with one Avro schema, then 500 with a second schema starting at id 500, and expects all three clusters to receive every message.

Why this is the same bug this PR fixes

It is the same readEntriesComplete() skip-branch stall, reached through the schema-fetch rewind rather than the failed-publish rewind described in the PR motivation:

1. At the schema boundary, GeoPersistentReplicator.replicateEntries() rewinds the cursor to re-read with the new schema: beforeTerminateOrCursorRewinding(ReasonOfWaitForCursorRewinding.Fetching_Schema) (GeoPersistentReplicator.java:270), and after the schema is fetched, doRewindCursor(true) (GeoPersistentReplicator.java:286).
2. If a cursor read was already dispatched at that moment (made naturally reachable by the read pipelining in [fix][broker] Fix replicator getting stuck under rate limiter throttling and honor readBatchSize/maxReadSizeBytes on the default read path #26005), cursor.cancelPendingReadRequest() returns false, so cancelPendingReadTasks(false) only flags the in-flight task (skipReadResultDueToCursorRewind = true) and leaves it with entries == null — it does not complete it.
3. When that dispatched read later completes, the readEntriesComplete() skip branch discards the result without calling setEntries(...), so the task stays entries == null forever → hasPendingRead() stays true and getPermitsIfNoPendingRead() returns 0. The readMoreEntries() issued by doRewindCursor(true) then early-returns, so replication never resumes past the schema boundary. The remote consumer times out and the assertion fails.

So Fetching_Schema is a second trigger of the same root cause as the Failed_Publishing path in the PR description; because the fix is in the shared readEntriesComplete() skip branch, it covers both rewind reasons. Consistent with that, Broker Group 5 (which runs testReplicationWithSchema) is green on this PR's latest CI run.

[lhotari]

Why #26005 makes this stall much more likely (and why testReplicationWithSchema now fails almost every run)

The stall requires a cursor read to be in flight (entries == null, flagged-but-not-cancellable) at the moment the schema-fetch rewind runs synchronously inside GeoPersistentReplicator.replicateEntries (beforeTerminateOrCursorRewinding(Fetching_Schema)). How often that coincidence happens is governed by how the replicator sizes and paces its reads — which #26005 changed.

Before #26005, the default read path (no dispatch rate limiter — the case in this test) read getPermitsIfNoPendingRead() entries, i.e. up to replicationProducerQueueSize (default 1000), with byteSize = -1 (unbounded), and did not apply readBatchSize:

// pre-#26005 readMoreEntries(), rate-limiter-disabled branch:
cursor.asyncReadEntriesOrWait(newInFlightTask.readingEntries /* up to producerQueueSize */, -1, ...);

So a replicator draining this test's 1000-message backlog requested it in essentially one read. The PersonOne → PersonThree schema boundary (msg ~500) fell inside that single batch, leaving little opportunity for a separate read to be in flight when the rewind fired — consistent with the stall being latent/rare before #26005.

After #26005 ("honor readBatchSize/maxReadSizeBytes on the default read path"), every read now goes through getReadLimits(permits), which caps it to min(permits, readBatchSize) messages and readMaxSizeBytes. By default readBatchSize = min(replicationProducerQueueSize, dispatcherMaxReadBatchSize) = 100.

(Worth noting for the record: readBatchSize does not start at 1 in steady state — it is initialized to getMaxReadBatchSize() = 100 in the constructor. It only drops to dispatcherMinReadBatchSize = 1 and ramps back up by doubling after a read failure, in readEntriesFailed ("reduce read batch size to avoid flooding bookies with retries"). So on the normal backlog-drain path it is a constant 100.)

So the same backlog is now drained in ~10 reads of 100, continuously pipelined — each send-completion frees producer-queue permits and re-triggers readMoreEntries while the current batch is still replicating. With ~10× more, smaller, pipelined reads, a cursor read is in flight far more often when replicateEntries reaches the schema boundary, and that in-flight read is exactly what the Fetching_Schema rewind strands.

This matches CI: on run 28466180778 (a branch based on master without this fix), testReplicationWithSchema failed in all three re-run attempts — i.e. it is now near-deterministic rather than an occasional flake.

So #26005 did not introduce this bug, but by bounding the default read path to readBatchSize it turned a latent/rare stall into a frequent one. This fix is effectively resolving that frequency regression, which argues for merging it promptly.

[lhotari]

Merging this PR since CI fails due to the bug that this PR fixes. The bug started appearing more often after PR #26005 as explained in the previous comment.