Move realtime stream offset fetch out of the ideal-state update lock by shounakmk219 · Pull Request #19170 · apache/pinot

shounakmk219 · 2026-08-06T06:02:38Z

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).

RealtimeSegmentValidationManager's ensureAllPartitionsConsuming fetched the stream offsets inside the Helix ideal-state update lambda, so on a table with many partitions the offset I/O (which can take minutes) was held under the per-table ideal-state lock and re-run on every ZK CAS retry, stalling concurrent segment commits. Pre-fetch the offsets from a read-only snapshot of the ideal state, outside the lock (preFetchOffsets), and pass them into the package-private ensureAllPartitionsConsuming, whose updater lambda now performs only in-memory ideal-state mutation. Lock hold-time is proportional to the mutation, and the offset fetch runs once regardless of CAS retries. The smallest-offset stream fetch is gated (anyPartitionNeedsSmallestOffset) so it runs only on a reset or when a partition actually needs a new CONSUMING segment. A partition that starts needing repair after the lock-free snapshot is deferred to the next validation run rather than being repaired with substituted start offsets.

codecov-commenter · 2026-08-06T06:56:23Z

Codecov Report

❌ Patch coverage is 79.68750% with 13 lines in your changes missing coverage. Please review.
✅ Project coverage is 67.09%. Comparing base (899e40d) to head (cc79c93).
⚠️ Report is 51 commits behind head on master.

Files with missing lines	Patch %	Lines
.../core/realtime/PinotLLCRealtimeSegmentManager.java	79.68%	6 Missing and 7 partials ⚠️

Additional details and impacted files

@@             Coverage Diff              @@
##             master   #19170      +/-   ##
============================================
- Coverage     67.10%   67.09%   -0.02%     
  Complexity     1424     1424              
============================================
  Files          3468     3468              
  Lines        222353   222377      +24     
  Branches      34999    35002       +3     
============================================
- Hits         149212   149205       -7     
- Misses        61224    61266      +42     
+ Partials      11917    11906      -11

Flag	Coverage Δ
integration	`100.00% <ø> (ø)`
integration1	`100.00% <ø> (ø)`
integration2	`0.00% <ø> (ø)`
java-25	`67.09% <79.68%> (-0.02%)`	⬇️
lane-a	`100.00% <ø> (ø)`
lane-b	`0.00% <ø> (ø)`
temurin	`67.09% <79.68%> (-0.02%)`	⬇️
unittests	`67.09% <79.68%> (-0.02%)`	⬇️
unittests1	`57.87% <ø> (ø)`
unittests2	`38.83% <79.68%> (-0.01%)`	⬇️

Flags with carried forward coverage won't be shown. Click here to find out more.

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

…-fix

Jackie-Jiang · 2026-08-11T23:49:56Z

+    // stream round-trips can take minutes; doing them here (against a snapshot of the ideal state) keeps the
+    // per-table ideal-state lock hold-time proportional to the in-memory ideal-state mutation, not to the offset
+    // I/O. The updater lambda below performs no stream I/O, so it is also cheap to re-run on ZK CAS retries.
+    IdealState snapshotIdealState = HelixHelper.getTableIdealState(_helixManager, realtimeTableName);


Are we fetching the latest consuming segment offset here? If so, we will lose recent segment commit when using IS snapshot right?

only the offset pre-fetch inputs come from the snapshot, every read that drives a repair decision or a written offset is from the fresh IS under the lock. We fetch the latestSegmentZKMetadataMap again as part of ensureAllPartitionsConsuming.

(MAJOR) I'm still confused. _streamMetadataList is provided by the PreFetchedOffsets, and I don't see it being updated during the retry. How does it handle new segment being added?

the _streamMetadataList answers only "which stream partitions exist and what's their start offset," and a segment commit doesn't change that, for retries we anyways get fresh IS and fresh ZK read (latestSegmentZKMetadataMap) which are used for any mutation decisions.

Please double check what is stored within the StreamMetadata. Based on my code reading, the PartitionGroupMetadata stores the start offset of the next consuming segment (Pinot knowledge), instead of the oldest offset available from upstream.

Copilot

Pull request overview

Moves realtime stream offset fetching outside the IdealState update lock to reduce lock contention and avoid repeated fetches on CAS retries.

Changes:

Prefetches stream metadata and smallest offsets before IdealState updates.
Defers repairs when required offsets were not prefetched.
Adds regression tests for offset handling and restoration.

Reviewed changes

Copilot reviewed 2 out of 2 changed files in this pull request and generated 3 comments.

File	Description
`PinotLLCRealtimeSegmentManager.java`	Implements off-lock offset prefetching and repair gating.
`PinotLLCRealtimeSegmentManagerTest.java`	Adds tests for prefetched offsets and deferred repairs.

💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.

shounakmk219 · 2026-08-17T05:24:53Z

        boolean isTableEnabled = idealState.isEnabled();
        boolean isTablePaused = isTablePaused(idealState);
-        boolean offsetsHaveToChange = offsetCriteria != null;
        if (isTableEnabled && !isTablePaused) {
-          List<PartitionGroupConsumptionStatus> currentPartitionGroupConsumptionStatusList =
-              offsetsHaveToChange ? List.of()
-                  // offsets from metadata are not valid anymore; fetch for all partitions
-                  : getPartitionGroupConsumptionStatusList(idealState, streamConfigs);
-          // FIXME: Right now, we assume topics are sharing same offset criteria
-          OffsetCriteria originalOffsetCriteria = streamConfigs.get(0).getOffsetCriteria();
-          // Read the smallest offset when a new partition is detected
-          streamConfigs.stream()
-              .forEach(streamConfig -> streamConfig.setOffsetCriteria(
-                  offsetsHaveToChange ? offsetCriteria : OffsetCriteria.SMALLEST_OFFSET_CRITERIA));
-          List<StreamMetadata> streamMetadataList =
-              getNewStreamMetadataList(streamConfigs, currentPartitionGroupConsumptionStatusList, idealState);
-          streamConfigs.stream().forEach(streamConfig -> streamConfig.setOffsetCriteria(originalOffsetCriteria));
-          return ensureAllPartitionsConsuming(tableConfig, streamConfigs, idealState, streamMetadataList,
-              offsetCriteria);
+          return ensureAllPartitionsConsuming(tableConfig, streamConfigs, idealState,
+              preFetchedOffsets._streamMetadataList, offsetCriteria,


topic level pause was not handled before so skipping it in this refactor as well

…nt lookup Addresses review feedback on the offset pre-fetch: anyPartitionNeedsSmallestOffset previously built a full latest-segment ZK metadata map just to decide whether the smallest-offset fetch was needed, adding a third O(partitions) PropertyStore scan on every healthy validation cycle (on top of getPartitionGroupConsumptionStatusList and the updater's own read). The gate now derives the latest LLC segment per partition from the snapshot ideal state's segment names alone (no ZK reads) and only builds the latest-segment ZK metadata map when the smallest-offset fetch is actually required. Extract the repeated "latest LLC segment per partition" logic into getLatestLLCSegmentPerPartition(Collection<String>) and reuse it in getLatestSegmentZKMetadataMap, getPartitionGroupConsumptionStatusList, and the gate.

…e update The existing test helper called preFetchOffsets and the package-private repair sequentially, so no test covered the public ensureAllPartitionsConsuming path's core guarantee. Add a test that drives the public method with a controlled static HelixHelper: it supplies the snapshot ideal state and applies the updater twice to simulate a ZK CAS conflict, then asserts the stream offset fetch happened before HelixHelper.updateIdealState (outside the updater) and was not repeated on retry.

…-fix

Jackie-Jiang · 2026-08-21T19:15:17Z

+    // stream round-trips can take minutes; doing them here (against a snapshot of the ideal state) keeps the
+    // per-table ideal-state lock hold-time proportional to the in-memory ideal-state mutation, not to the offset
+    // I/O. The updater lambda below performs no stream I/O, so it is also cheap to re-run on ZK CAS retries.
+    IdealState snapshotIdealState = HelixHelper.getTableIdealState(_helixManager, realtimeTableName);


(MAJOR) I'm still confused. _streamMetadataList is provided by the PreFetchedOffsets, and I don't see it being updated during the retry. How does it handle new segment being added?

…-fix

Minor cleanup in the latest-LLC-segment helpers: rename the local map in getLatestLLCSegmentPerPartition, and pre-size the result map in getLatestSegmentZKMetadataMap with Maps.newHashMapWithExpectedSize since its final size equals the number of latest segments.

noob-se7en · 2026-08-26T13:20:08Z

          if (partitionIdToSmallestOffset == null) {
-            partitionIdToSmallestOffset =
-                fetchPartitionGroupIdToSmallestOffset(streamConfigs, idealState, latestSegmentZKMetadataMap);
+            LOGGER.info("Smallest stream offsets not fetched this cycle; deferring repair of partition: {} of table: "


can this happen? If yes, then why not fetchPartitionGroupIdToSmallestOffset here as fallback?

this can happen if a partition looses its consuming segment between the snapshot and the lock or retries on updater. Not falling back to fetchPartitionGroupIdToSmallestOffset as this PR is trying to pull out all the IO ops out of the IS lock, it should be picked in the next RVM run anyways.

noob-se7en · 2026-08-26T13:22:26Z

-                latestSegmentName);
+          // Do not create a new CONSUMING segment when the partition has no smallest stream offset (it has reached
+          // end of life).
+          StreamPartitionMsgOffset smallestStreamOffset = partitionIdToSmallestOffset.get(partitionId);


Is it guaranteed partitionIdToSmallestOffset will have partitionId?

yes it should have all the partitionId entries fetched from upstream

noob-se7en · 2026-08-27T12:16:44Z

+  private boolean anyPartitionNeedsSmallestOffset(IdealState idealState) {
+    Map<String, Map<String, String>> instanceStatesMap = idealState.getRecord().getMapFields();
+    Map<Integer, LLCSegmentName> partitionGroupIdToLatestSegment =
+        getLatestLLCSegmentPerPartition(instanceStatesMap.keySet());


this derives the latest segment from IS names but the repair loop derives it from getLLCSegments (property store), so a CONSUMING segment in IS with no ZK metadata (the orphan IdealStateGroupCommit's cancellation comment describes) keeps the gate false on every run while the loop keeps deferring that partition, and unlike pre-PR nothing heals it (repairSegmentsInErrorState skips metadata-less segments). gating on getLatestLLCSegmentPerPartition(getLLCSegments(realtimeTableName)) against the snapshot IS is one getChildNames call and matches the loop exactly.

noob-se7en · 2026-08-27T12:16:44Z

+    //   - null otherwise: the lock-free snapshot gate saw no partition needing a new CONSUMING segment, so the
+    //     smallest offsets were not fetched. Start offsets are NOT the stream-smallest in this case, so they must
+    //     not be substituted; a partition that turns out to need a new segment now (it started needing repair after
+    //     the snapshot) is deferred to the next validation run below.


"next validation run" is one RVM period, 1h by default, not the 15 min in the description, and the window that lands a partition here is the whole pre-fetch. pre-PR a last replica going OFFLINE mid-fetch was still repaired in the same run via the CAS retry; worth a forced re-run when the lambda defers, or at least the 1h in the description.

shounakmk219 added ingestion Related to data ingestion pipeline performance Related to performance optimization real-time Related to realtime table ingestion and serving labels Aug 6, 2026

Merge remote-tracking branch 'upstream/master' into rvm-is-lock-stall…

c011119

…-fix

Jackie-Jiang requested a balanced review from Copilot August 11, 2026 23:47

Copilot started reviewing on behalf of Jackie-Jiang August 11, 2026 23:47 View session

Jackie-Jiang reviewed Aug 11, 2026

View reviewed changes

Copilot AI reviewed Aug 11, 2026

View reviewed changes

shounakmk219 added 3 commits August 17, 2026 13:11

Merge remote-tracking branch 'upstream/master' into rvm-is-lock-stall…

588cfae

…-fix

Jackie-Jiang reviewed Aug 21, 2026

View reviewed changes

shounakmk219 added 3 commits August 24, 2026 13:36

convert PreFetchedOffsets into a record

4b21bb5

Merge remote-tracking branch 'upstream/master' into rvm-is-lock-stall…

6dc9f2e

…-fix

noob-se7en reviewed Aug 26, 2026

View reviewed changes

noob-se7en reviewed Aug 27, 2026

View reviewed changes

Uh oh!

Conversation

shounakmk219 commented Aug 6, 2026

Problem

Fix

Testing

Concurrency notes (IS updated by another process during an RVM run)

Depends on

Uh oh!

codecov-commenter commented Aug 6, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Codecov Report

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Reviewed changes

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

5 participants

codecov-commenter commented Aug 6, 2026 •

edited

Loading