[fix][broker] Handle synchronous schema lookup failures in replication

[Denovo1998]

Motivation

Geo replication pauses and rewinds the cursor when schema information for a replicated message is not immediately available.

However, getSchemaInfo(msg) can also fail synchronously before returning a CompletableFuture. The underlying schema lookup can throw ExecutionException, a RuntimeException such as Guava's UncheckedExecutionException, or ExecutionError.

This lookup happens after headersAndPayload.retain(). Previously, a synchronous failure bypassed the existing schema-fetch cleanup and cursor-rewind path. As a result, the current entry and retained payload reference could remain unreleased, the deserialized message would not be recycled, and the corresponding in-flight entry would not be marked as completed.

Modifications

• Normalize synchronous schema lookup failures into failed CompletableFutures so they follow the existing schema-fetch failure handling path.
• Ensure resources and in-flight state are cleaned up correctly when schema lookup fails synchronously.
• Add MESSAGE_RATE_BACKOFF_MS before retrying after a failed schema fetch to avoid a tight retry loop.
• Keep cursor-read scheduling coordinated through the existing rewind state handling in readMoreEntries().
• Add regression tests covering checked, unchecked, and Guava-wrapped synchronous schema lookup failures.

Verifying this change

The regression test was verified locally with:

./gradlew :pulsar-broker:test \
  --tests 'org.apache.pulsar.broker.service.persistent.GeoPersistentReplicatorTest' \
  -PexcludedTestGroups=''

GitHub CI is currently pending workflow approval.

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]

Thanks for the fix. The cleanup path makes sense to me.

I left a few comments around exception handling, retry behavior, and test coverage.

[void-ptr974]

Would it be better to narrow this catch to the expected exception type? Since getSchemaInfo only declares ExecutionException, catching all Exceptions could accidentally turn unrelated bugs into schema retry loops. Another option might be to normalize getSchemaInfo to return a failed future and then reuse the existing schemaFuture.isCompletedExceptionally() path.

[Denovo1998]

Good point. I narrowed the catch to ExecutionException and normalize that synchronous schema lookup failure into a failed future, so unexpected exceptions are no longer converted into schema retry loops while the existing schema future cleanup path is reused.

[void-ptr974]

This path can immediately rewind and re-read the same entry if the synchronous schema lookup failure persists. replicateEntries() returns false, so readEntriesComplete() may call readMoreEntries() right away.

One way to avoid a tight retry loop is to keep the replicator in the cursor-rewinding wait state and schedule doRewindCursor(true) after a small backoff instead of rewinding immediately.

[Denovo1998]

Agreed. The exceptional schema future path now keeps the replicator in the cursor-rewinding wait state and schedules doRewindCursor(true) after MESSAGE_RATE_BACKOFF_MS. Successful schema fetches still rewind immediately.

[void-ptr974]

This test only covers the current entry cleanup. It does not verify the batch behavior after skipRemainingMessages is set.

Please extend it to use a multi-entry batch and verify that the remaining entries are skipped/released, completedEntries reaches the full batch size, and the cursor is rewound.

[Denovo1998]

Extended the regression test to use a multi-entry batch. It now verifies the remaining entry is skipped and released, completedEntries reaches the full batch size, and cursor rewind/read retry is triggered only by the scheduled backoff task.

[void-ptr974]

Thanks for the update. The main concerns look addressed.

One small follow-up: after this path calls beforeTerminateOrCursorRewinding(...), replicateEntries() returns false, so readEntriesComplete() will still call readMoreEntries(). That call should not start a read while waitForCursorRewindingRefCnf > 0, but it may schedule another delayed retry. Could we skip the outer readMoreEntries() when the replicator is already waiting for cursor rewind, and let the scheduled doRewindCursor(true) resume reads instead?

[Denovo1998]

Thanks for the update. The main concerns look addressed.

One small follow-up: after this path calls beforeTerminateOrCursorRewinding(...), replicateEntries() returns false, so readEntriesComplete() will still call readMoreEntries(). That call should not start a read while waitForCursorRewindingRefCnf > 0, but it may schedule another delayed retry. Could we skip the outer readMoreEntries() when the replicator is already waiting for cursor rewind, and let the scheduled doRewindCursor(true) resume reads instead?

I added a guard in readEntriesComplete() to skip the outer readMoreEntries() while the replicator is already waiting for cursor rewind. This leaves the scheduled doRewindCursor(true) as the path that resumes reads. I also updated the regression test to exercise readEntriesComplete() end-to-end and verify no extra read is triggered before the scheduled rewind runs.

[void-ptr974]

LGTM. Thanks for addressing the comments.

[lhotari]

I ran an AI-assisted review of this PR (combined Claude Fable 5 + OpenAI Codex gpt-5.6-sol review; findings merged and verified against the code). Overall the fix looks real and correctly targeted: getSchemaInfo() is a Guava LoadingCache.get() call that throws ExecutionException synchronously, and in current master that throw lands in the outer catch (Exception) after headersAndPayload.retain() — leaking the retained buffer, the entry, the MessageImpl and the in-flight permit, with the cursor neither rewound nor resumed. Routing the failure into the existing isCompletedExceptionally() skip-path reuses the proven cleanup + rewind machinery, and the new readEntriesComplete() guard properly defers read resumption to the scheduled doRewindCursor(true).

Findings, in decreasing severity:

1. Catching only ExecutionException leaves the same leak for unchecked synchronous failures (GeoPersistentReplicator.replicateEntries). Guava's LoadingCache.get() also throws UncheckedExecutionException (loader threw a RuntimeException) and ExecutionError, and getSchemaByVersion() can throw unchecked synchronously. Any of those still escape to the outer catch (Exception e), which only logs — reproducing exactly the failure mode this PR sets out to fix. Suggest broadening to catch (Exception e), or cleaner: move the try/catch into getSchemaInfo() so it returns a failed future and drop throws ExecutionException (a CompletableFuture-returning method shouldn't throw synchronously; GeoPersistentReplicator is its only caller and ShadowReplicator doesn't use it, so the signature change is contained).

2. The regression test self-repairs the leak it's meant to detect (GeoPersistentReplicatorTest, the finally block). The loop releasing headersAndPayload until refCnt() == 0 means the test would still pass if the production path forgot headersAndPayload.release(). Suggest asserting assertThat(headersAndPayload.refCnt()).isZero() right after the verifications, before any fallback cleanup. If finding 1 is addressed, please also add a companion test injecting an unchecked exception (e.g. UncheckedExecutionException), which the current test cannot cover.

3. Two behavior changes are not reflected in the PR description. The diff also (a) adds a MESSAGE_RATE_BACKOFF_MS (1s) delay before doRewindCursor(true) for all schema-fetch failures, including asynchronous ones — previously an async failure rewound immediately, so a persistently failing schema fetch could hot-loop read → fail → rewind → re-read; and (b) replaces the old 1s scheduled-retry polling from readEntriesComplete() with an explicit hand-off to the scheduled rewind. Both are good changes, but the description/commit message should state them — especially with the release/4.0.13 and release/4.2.4 labels, since backporters need the full behavioral delta (and should verify those branches have the InFlightTask/waitForCursorRewindingRefCnf structure this patch assumes).

4. Minor: the new debug log in readEntriesComplete() reads reasonOfWaitForCursorRewinding, which is non-volatile and written under the inFlightTasks lock, so the log line can print a stale or null reason. Harmless (log-only), just confirming it's intentional.

5. Minor: the scheduled rewind is fire-and-forget — if the broker executor rejects the task at shutdown, the exception is swallowed inside whenComplete and waitForCursorRewindingRefCnf never decrements, stalling that replicator until unload. This matches the existing idiom in readMoreEntries(), so it's acceptable as-is.

Concurrency was checked independently by both reviews and no race was found in the new guard: (a) for Fetching_Schema, resume is deterministically owned by doRewindCursor(true) (immediate on success, backoff-scheduled on failure), and if the rewind wins the race against the guard, the fallthrough readMoreEntries() safely no-ops on hasPendingRead(); (b) for the Failed_Publishing transient window (refcount briefly > 0 between beforeTerminateOrCursorRewinding and doRewindCursor(false) on the producer thread), resumption is still guaranteed because the same sendComplete continues on its thread and its queue-drain logic calls readMoreEntries() after the refcount is back to 0; (c) Terminating is handled by the earlier state check in readEntriesComplete().

[Denovo1998]

I ran an AI-assisted review of this PR (combined Claude Fable 5 + OpenAI Codex gpt-5.6-sol review; findings merged and verified against the code). Overall the fix looks real and correctly targeted: getSchemaInfo() is a Guava LoadingCache.get() call that throws ExecutionException synchronously, and in current master that throw lands in the outer catch (Exception) after headersAndPayload.retain() — leaking the retained buffer, the entry, the MessageImpl and the in-flight permit, with the cursor neither rewound nor resumed. Routing the failure into the existing isCompletedExceptionally() skip-path reuses the proven cleanup + rewind machinery, and the new readEntriesComplete() guard properly defers read resumption to the scheduled doRewindCursor(true).

Findings, in decreasing severity:

1. Catching only ExecutionException leaves the same leak for unchecked synchronous failures (GeoPersistentReplicator.replicateEntries). Guava's LoadingCache.get() also throws UncheckedExecutionException (loader threw a RuntimeException) and ExecutionError, and getSchemaByVersion() can throw unchecked synchronously. Any of those still escape to the outer catch (Exception e), which only logs — reproducing exactly the failure mode this PR sets out to fix. Suggest broadening to catch (Exception e), or cleaner: move the try/catch into getSchemaInfo() so it returns a failed future and drop throws ExecutionException (a CompletableFuture-returning method shouldn't throw synchronously; GeoPersistentReplicator is its only caller and ShadowReplicator doesn't use it, so the signature change is contained).
2. The regression test self-repairs the leak it's meant to detect (GeoPersistentReplicatorTest, the finally block). The loop releasing headersAndPayload until refCnt() == 0 means the test would still pass if the production path forgot headersAndPayload.release(). Suggest asserting assertThat(headersAndPayload.refCnt()).isZero() right after the verifications, before any fallback cleanup. If finding 1 is addressed, please also add a companion test injecting an unchecked exception (e.g. UncheckedExecutionException), which the current test cannot cover.
3. Two behavior changes are not reflected in the PR description. The diff also (a) adds a MESSAGE_RATE_BACKOFF_MS (1s) delay before doRewindCursor(true) for all schema-fetch failures, including asynchronous ones — previously an async failure rewound immediately, so a persistently failing schema fetch could hot-loop read → fail → rewind → re-read; and (b) replaces the old 1s scheduled-retry polling from readEntriesComplete() with an explicit hand-off to the scheduled rewind. Both are good changes, but the description/commit message should state them — especially with the release/4.0.13 and release/4.2.4 labels, since backporters need the full behavioral delta (and should verify those branches have the InFlightTask/waitForCursorRewindingRefCnf structure this patch assumes).
4. Minor: the new debug log in readEntriesComplete() reads reasonOfWaitForCursorRewinding, which is non-volatile and written under the inFlightTasks lock, so the log line can print a stale or null reason. Harmless (log-only), just confirming it's intentional.
5. Minor: the scheduled rewind is fire-and-forget — if the broker executor rejects the task at shutdown, the exception is swallowed inside whenComplete and waitForCursorRewindingRefCnf never decrements, stalling that replicator until unload. This matches the existing idiom in readMoreEntries(), so it's acceptable as-is.

Concurrency was checked independently by both reviews and no race was found in the new guard: (a) for Fetching_Schema, resume is deterministically owned by doRewindCursor(true) (immediate on success, backoff-scheduled on failure), and if the rewind wins the race against the guard, the fallthrough readMoreEntries() safely no-ops on hasPendingRead(); (b) for the Failed_Publishing transient window (refcount briefly > 0 between beforeTerminateOrCursorRewinding and doRewindCursor(false) on the producer thread), resumption is still guaranteed because the same sendComplete continues on its thread and its queue-drain logic calls readMoreEntries() after the refcount is back to 0; (c) Terminating is handled by the earlier state check in readEntriesComplete().

@lhotari
Okay, I've reviewed it again and made some changes. Take a look once more.

[void-ptr974]

LGTM

Dropping approval while checking for possible race conditions.

[lhotari]

Thanks for the follow-up — this is what I was asking for.

a6b5d111 drops the waitForCursorRewindingRefCnf branch from readEntriesComplete(), so readMoreEntries() is now the single place that consults the rewind state, and it does so at PersistentReplicator.java:310 inside synchronized (inFlightTasks). Falling through to readMoreEntries() is safe precisely because that method re-checks under the lock and self-skips. Dropping the readMoreEntries() override in the test is a real improvement too: the test now drives the actual read-scheduling path and asserts the externally observable behavior (cursor.asyncReadEntriesOrWait(...) never before the rewind, exactly once after) rather than an internal call count.

I checked the CompletableFuture contract on this path end to end, since that is the part that matters most here:

• getSchemaInfo() now has its entire body inside the try, so there is no statement left that can throw before a future exists.
• The ExecutionException | RuntimeException | ExecutionError catch list is an exact match for Guava's documented LoadingCache.get(K) contract (checked → ExecutionException, unchecked → UncheckedExecutionException, error → ExecutionError), and both caches on this path are Guava LoadingCaches — PulsarClientImpl.schemaProviderLoadingCache (PulsarClientImpl.java:200) and the per-topic cache inside MultiVersionSchemaInfoProvider. It deliberately does not swallow bare Error, which is right.
• The failure is surfaced, not swallowed: log.warn().exception(e) at GeoPersistentReplicator.java:282-285.
• The refcount stays balanced. skipRemainingMessages guarantees at most one beforeTerminateOrCursorRewinding() per replicateEntries() call, and both whenComplete branches reach doRewindCursor(true) exactly once, so a persistently failing lookup settles into a ~1s retry cycle and recovers if the lookup ever succeeds — it does not strand the replicator. The added backoff also removes what was previously an unthrottled rewind→read→fail loop on the failure path.

I ran the new test locally and also re-ran it with only the getSchemaInfo() change reverted, to confirm it actually pins the fix:

./gradlew :pulsar-broker:test --tests 'org.apache.pulsar.broker.service.persistent.GeoPersistentReplicatorTest' -PexcludedTestGroups=''

At a6b5d111: 3 tests, 0 failures. With getSchemaInfo() reverted to the throwing version: fails at GeoPersistentReplicatorTest.java:95 with expected: 2 but was: 0 — the in-flight permits are never released. That is the bug, reproduced.

Two non-blocking notes, plus one small logging nit inline:

• CI has never run on this PR. Both Pulsar CI and Pulsar CI Flaky on a6b5d111 are sitting at action_required since 2026-08-03, so there are zero check runs on the head commit. That is the workflow-approval gate for contributor PRs, not anything you did wrong — I need to approve the runs. I would rather see a green CI before approving a change in the replication read path, so I am leaving this as a comment for now.
• The last bullet of the PR description ("Skip the outer readMoreEntries() call while waiting for cursor rewind, leaving doRewindCursor(true) responsible for resuming reads") now describes the code that a6b5d111 removed. Since we squash-merge the description into the commit message, it is worth updating that bullet before merge.

[lhotari]

[NIT] the failure path logs "schema fetching done" at INFO once per second while the fetch is still failing

This line is reused verbatim from the success branch at line 292, but here the schema fetch did not complete — we are backing off and retrying after the lookup failed.

During a sustained schema-lookup outage the replicator settles into a ~1s cycle, so the broker log gets an INFO Resume the data replication after the schema fetching done every second while replication is in fact making no progress. Someone reading the log to diagnose a stalled replicator would be told the opposite of what is happening, and it partly cancels out the WARN you added just above.

Something like Retrying the data replication after a failed schema fetch would read correctly, and per CODING.md a periodic retry line like this is probably better at DEBUG than INFO.

[Denovo1998]

done.