shounakmk219 opened a new pull request, #19170: URL: https://github.com/apache/pinot/pull/19170
## Problem `RealtimeSegmentValidationManager` (RVM) calls `PinotLLCRealtimeSegmentManager.ensureAllPartitionsConsuming`, which fetched the stream offsets **inside** the Helix ideal-state update lambda (`HelixHelper.updateIdealState` → `IdealStateGroupCommit`). On a table with many partitions the offset I/O can take minutes, and because it ran inside the updater it (a) held the per-table ideal-state lock for that whole time — stalling concurrent segment commits — and (b) was re-executed on every ZK version-checked CAS retry. This is the follow-up to #19116 (which batched the Kafka fetch): batching shrinks the fetch, this change removes it from the lock entirely. ## Fix - **Pre-fetch outside the lock.** `ensureAllPartitionsConsuming` now reads a read-only snapshot of the ideal state (`HelixHelper.getTableIdealState`), does the early enabled/paused check, and calls a new `@VisibleForTesting preFetchOffsets(...)` to compute `streamMetadataList` and the per-partition smallest offsets **before** entering `HelixHelper.updateIdealState`. - **Lambda does in-memory work only.** The updater lambda re-checks enabled/paused on the fresh IS and calls the package-private `ensureAllPartitionsConsuming(..., preFetchedPartitionIdToSmallestOffset)`, which mutates the fresh IS from the pre-fetched offsets. No stream I/O runs under the lock, so lock hold-time is proportional to the mutation and the fetch is not repeated on CAS retries. - **Gated smallest-offset fetch.** The smallest-offset round-trip only happens on a reset (`offsetCriteria != null`) or when `anyPartitionNeedsSmallestOffset(...)` finds a partition whose latest segment has no CONSUMING replica (the only repair path that consults it). Healthy tables fetch nothing. - **Snapshot/fresh divergence is safe.** The IS mutation is still fresh + version-checked CAS, so no concurrent update is lost. Offsets are the only snapshot input; the three-way smallest-offset fallback plus a null-safe skip guard mean a partition that starts needing repair *after* the snapshot is deferred to the next run rather than being repaired with a substituted (checkpoint) offset. The actual start offset written in the periodic path comes from fresh ZK segment metadata. ## Testing - `PinotLLCRealtimeSegmentManagerTest` + `RealtimeSegmentValidationManagerTest`: 57 tests pass, including new cases — `testEnsureAllPartitionsConsumingHonorsPreFetchedSmallestOffset`, `testEnsureAllPartitionsConsumingDefersRepairWhenSmallestOffsetsNotPreFetched`, `testPreFetchOffsetsSkipsSmallestOffsetFetchForHealthyTable`, and `testPreFetchOffsetsRestoresOffsetCriteriaOnFailure`. - spotless / checkstyle / license clean. ## Concurrency notes (IS updated by another process during an RVM run) The mutation path is unchanged (serialized per table via `IdealStateGroupCommit`, version-checked CAS), so concurrent commits are not lost — a commit landing mid-run causes a CAS retry and RVM re-runs on the newer IS. Only the offset inputs are a snapshot. Residual, all self-healing within one 15-min cycle: - A partition that loses its CONSUMING replica between snapshot and lock is deferred one cycle (not repaired with a wrong offset). - A stale-low smallest offset can only under-report data loss for one cycle (never over-report); a truncated-stream segment briefly goes OFFLINE and is re-repaired next cycle with a fresh smallest. - New stream partitions / unpause are picked up on the next cycle. ## Depends on Complementary to #19116 (batching). Independent at the code level — this PR only touches `pinot-controller` and calls pre-existing stream methods — but both together give the full win (fast fetch + off-lock). -- 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] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
