Denovo1998 commented on PR #26108: URL: https://github.com/apache/pulsar/pull/26108#issuecomment-5081145469
> 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. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
