You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pinot.apache.org by "suvodeep-pyne (via GitHub)" <gi...@apache.org> on 2026/10/21 19:50:00 UTC
[PR] Remove the kept new segment when a resumed pauseless commit start completes without it [pinot]
suvodeep-pyne opened a new pull request, #19995:
URL: https://github.com/apache/pinot/pull/19995
## Summary
Follow-up to #19740 for [Jackie's post-merge review](https://github.com/apache/pinot/pull/19740#discussion_r4331476685). When the IdealState update of a pauseless commit start fails, #19740 keeps the new segment's ZK metadata for the next attempt. If the table is paused before that attempt (or the topic is paused, or the partition leaves the stream), the retry completes the commit without a new segment, but the kept segment stays. `RealtimeSegmentValidationManager` later takes it as the latest segment of the partition and adds it to the IdealState. For example, resuming the paused table with `consumeFrom=largest` then starts that partition from the committed end offset instead.
## Changes
- A resumed commit start looks for the earlier attempt's new segment before step 2 decides whether the partition still needs a new segment. `createNewSegmentMetadata` takes that segment instead of the `resumed` flag (Jackie's second suggestion). It uses the segment only after the existing checks pass, i.e. the partition is still in the stream and the table/topic is not paused, so finding it never by itself gets it assigned.
- After step 3, if the commit completed without the earlier attempt's segment and that segment is not in the IdealState, its ZK metadata is removed. The committing segment is ONLINE by then, so no attempt of the commit can add the segment anymore.
Not changed:
- If no retry succeeds before the max segment completion time, the kept segment stays, and the validation manager adds it, as it does after a controller stop between steps 2 and 3. Removing it in the failure handler while the table is paused could race with a resume and a resumed attempt on another controller, and leave a CONSUMING segment without ZK metadata.
- The new cleanup removes the metadata without a fence against the IdealState, like the existing cleanups in the same method. An attempt that stalls between its IdealState update and the removal could remove a segment of the same name that validation created in the meantime, which needs a resume and a repair within the same minute as the first attempt.
- Pauseless tables only support a single stream, and validation does not skip a paused topic of a single-stream table, so for topic pause the change only removes the kept segment.
- Jackie's first suggestion (`prepareCommitStart`) would change step 1 of every pauseless commit to save a few reads on the resume path. I kept it out of this fix.
## Testing
New tests in `PinotLLCRealtimeSegmentManagerTest`. Each one fails without the fix:
- The table is paused between the failed attempt and the retry. The retry removes the kept segment, and resuming the table with an offset criteria creates a new segment from the selected offset.
- The topic is paused between the failed attempt and the retry. The retry removes the kept segment.
- The partition is removed from the stream between the failed attempt and the retry. The retry completes without adding the kept segment and removes it.
Existing tests pass: `PinotLLCRealtimeSegmentManagerTest`, `SegmentCompletionTest`, `SegmentCompletionFSMFactoryTest`, `RealtimeSegmentValidationManagerTest`, and `PauselessRealtimeIngestionIntegrationTest`.
End to end, I ran a local cluster test (not part of this PR) with ZooKeeper, Kafka, a controller, a broker and a server. Every commit start fails before the IdealState update. The test then pauses one table and the topic of another, removes the failure so that the servers' retries succeed, and resumes the first table with `consumeFrom=largest`:
| | Kept segments after the retries | Resume from `largest` |
|---|---|---|
| Without the fix | Remain in both tables | Adds the kept segments, which start at the committed end offset (2500); the table ingests all 115545 rows |
| With the fix | Removed in both tables | New segments start at the end of the topic; the count stays at 5000 |
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Remove the kept new segment when a resumed pauseless commit start completes without it [pinot]
Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #19995:
URL: https://github.com/apache/pinot/pull/19995#issuecomment-6328223113
## [Codecov](https://app.codecov.io/gh/apache/pinot/pull/19995?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
:white_check_mark: All modified and coverable lines are covered by tests.
:white_check_mark: Project coverage is 100.00%. Comparing base ([`edc48df`](https://app.codecov.io/gh/apache/pinot/commit/edc48dff392553e812c516efb3ba890580e96965?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)) to head ([`9b00a71`](https://app.codecov.io/gh/apache/pinot/commit/9b00a71faf8c7955a11056012fd24283a81d7bbc?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)).
:warning: Report is 6 commits behind head on master.
<details><summary>Additional details and impacted files</summary>
```diff
@@ Coverage Diff @@
## master #19995 +/- ##
==============================================
+ Coverage 69.60% 100.00% +30.39%
+ Complexity 1491 6 -1485
==============================================
Files 3548 3 -3545
Lines 234805 6 -234799
Branches 37635 0 -37635
==============================================
- Hits 163434 6 -163428
+ Misses 58740 0 -58740
+ Partials 12631 0 -12631
```
| [Flag](https://app.codecov.io/gh/apache/pinot/pull/19995/flags?src=pr&el=flags&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [integration](https://app.codecov.io/gh/apache/pinot/pull/19995/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `100.00% <ø> (ø)` | |
| [integration1](https://app.codecov.io/gh/apache/pinot/pull/19995/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `100.00% <ø> (ø)` | |
| [integration2](https://app.codecov.io/gh/apache/pinot/pull/19995/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
| [java-25](https://app.codecov.io/gh/apache/pinot/pull/19995/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `100.00% <ø> (+30.39%)` | :arrow_up: |
| [lane-a](https://app.codecov.io/gh/apache/pinot/pull/19995/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `100.00% <ø> (ø)` | |
| [lane-b](https://app.codecov.io/gh/apache/pinot/pull/19995/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `0.00% <ø> (ø)` | |
| [temurin](https://app.codecov.io/gh/apache/pinot/pull/19995/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `100.00% <ø> (+30.39%)` | :arrow_up: |
| [unittests](https://app.codecov.io/gh/apache/pinot/pull/19995/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
| [unittests1](https://app.codecov.io/gh/apache/pinot/pull/19995/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
| [unittests2](https://app.codecov.io/gh/apache/pinot/pull/19995/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#carryforward-flags-in-the-pull-request-comment) to find out more.
</details>
[:umbrella: View full report in Codecov by Harness](https://app.codecov.io/gh/apache/pinot/pull/19995?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
<details><summary> :rocket: New features to boost your workflow: </summary>
- :package: [JS Bundle Analysis](https://docs.codecov.com/docs/javascript-bundle-analysis): Save yourself from yourself by tracking and limiting bundle sizes in JS merges.
</details>
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Remove the kept new segment when a resumed pauseless commit start completes without it [pinot]
Posted by "suvodeep-pyne (via GitHub)" <gi...@apache.org>.
suvodeep-pyne commented on PR #19995:
URL: https://github.com/apache/pinot/pull/19995#issuecomment-6354637100
After merging master (7abcc71b3f), the only failed check is Pinot Integration Test Set 1. It fails in `WindowFunnelTest.testFunnelCompleteCountGroupByQueries` (`expected [40] but found [36]`), a funnel query test that this controller-only change does not touch. The same check passed on the previous head, df8842da90. Could a committer re-run it?
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Remove the kept new segment when a resumed pauseless commit start completes without it [pinot]
Posted by "suvodeep-pyne (via GitHub)" <gi...@apache.org>.
suvodeep-pyne commented on code in PR #19995:
URL: https://github.com/apache/pinot/pull/19995#discussion_r4343161840
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManager.java:
##########
@@ -718,6 +722,15 @@ private void commitSegmentMetadataInternal(String realtimeTableName,
removeSegmentZKMetadataBestEffort(realtimeTableName, newConsumingSegmentName);
newConsumingSegmentName = null;
}
+ } else if (newSegmentOfEarlierAttempt != null
+ && !instanceStatesMapAfterStep3.containsKey(newSegmentOfEarlierAttempt)) {
+ // The committing segment is ONLINE now, so no attempt of the commit can add the new segment of the earlier
+ // attempt anymore. Remove it so that RealtimeSegmentValidationManager does not add it either.
+ LOGGER.info(
+ "Cleaning up segment ZK metadata for new segment {} of table {} created by an earlier attempt of the commit "
+ + "because it is not used. This can happen when table/topic consumption is paused, or the partition is "
+ + "no longer in the stream.", newSegmentOfEarlierAttempt, realtimeTableName);
+ removeSegmentZKMetadataBestEffort(realtimeTableName, newSegmentOfEarlierAttempt);
Review Comment:
Thanks, agreed, the snapshot-based delete was not safe. Fixed in f3c514c99f: the cleanup now goes through PinotHelixResourceManager#removeInProgressSegmentsZKMetadataNotInIdealState. It removes only IN_PROGRESS segments that are not in the IdealState, in one ZK multi() with version-checked deletes of their ZK metadata and a version-checked write of the unchanged IdealState:
- if validation read the IdealState before the removal, its write fails, and its retry reads the segment ZK metadata again and no longer finds the segment;
- if validation added the segment first, the IdealState version changed, so the removal fails and its retry keeps the segment.
Tests:
- testCommitStartRetryKeepsNewSegmentOfEarlierAttemptAddedByValidation: the retry blocks before the removal while the table is resumed and validation adds the kept segment; its ZK metadata is kept.
- Against a real ZooKeeper in PinotHelixResourceManagerStatelessTest: the segment added to the IdealState right after the removal read it is kept; its ZK metadata overwritten right after the read is kept; an IdealState write based on the IdealState read before the removal fails.
Removing either version check fails the matching test. The previous cleanup of a new segment not added to the IdealState (pause between steps 2 and 3) uses the same fenced removal.
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManager.java:
##########
@@ -673,9 +673,13 @@ private void commitSegmentMetadataInternal(String realtimeTableName,
// Step-2: Create new segment metadata if needed
long startTimeNs2 = System.nanoTime();
+ // Another attempt of the same commit start might have created the new segment already. Find it before deciding
+ // whether the partition still needs a new segment, so that it can be removed when it is not used.
+ String newSegmentOfEarlierAttempt =
+ resumed ? findNewSegmentOfEarlierAttempt(realtimeTableName, new LLCSegmentName(committingSegmentName)) : null;
Review Comment:
Fixed in f3c514c99f: the lookup now returns every new segment of earlier attempts in the window, not just the first. The commit reuses the oldest one when the partition still needs a new segment, and removes all the unused ones through the fenced removal. testCommitStartRetryWhileTablePausedRemovesNewSegmentsOfEarlierAttempts covers 2 kept segments from overlapping attempts; it fails if the lookup stops at the first one.
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Remove the kept new segment when a resumed pauseless commit start completes without it [pinot]
Posted by "suvodeep-pyne (via GitHub)" <gi...@apache.org>.
suvodeep-pyne commented on PR #19995:
URL: https://github.com/apache/pinot/pull/19995#issuecomment-6371865613
@Jackie-Jiang this addresses your post-merge review on #19740 and your two comments here, which are both resolved. xiangfu0 has approved, and all checks pass on c5eac9b531. Could you take a look, and merge if it looks good to you?
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Remove the kept new segment when a resumed pauseless commit start completes without it [pinot]
Posted by "xiangfu0 (via GitHub)" <gi...@apache.org>.
xiangfu0 commented on code in PR #19995:
URL: https://github.com/apache/pinot/pull/19995#discussion_r4343871352
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManager.java:
##########
@@ -2278,6 +2283,80 @@ private ZkClient getOrBuildMultiWriteZkClient() {
}
}
+ /// Removes the ZK metadata of the given segments that are IN_PROGRESS and not in the IdealState of the table, i.e.
+ /// new consuming segments that were not added to the IdealState, and returns the segments whose ZK metadata was
+ /// removed.
+ ///
+ /// The removal is submitted in one ZK `multi()` transaction with version-checked deletes of the segment ZK metadata
+ /// read here, and a version-checked write of the unchanged IdealState:
+ /// - An IdealState update that read the IdealState before the transaction fails its version check. An updater that
+ /// reads the segment ZK metadata again when it retries, like the one of RealtimeSegmentValidationManager, then no
+ /// longer finds the removed segments.
+ /// - If the IdealState, or the ZK metadata of one of the segments, is updated after it was read here, the transaction
+ /// fails and is retried.
+ ///
+ /// The data versions cannot tell a segment ZK metadata deleted and recreated with the same name since it was read,
+ /// e.g. by an upload under the same name, from the one that was read. As the IdealState is outside the property
+ /// store, the transaction uses the multi-write ZkClient directly instead of [ZkMultiWriteBuilder].
+ public Set<String> removeInProgressSegmentsZKMetadataNotInIdealState(String tableNameWithType,
+ Collection<String> segmentNames) {
+ return removeInProgressSegmentsZKMetadataNotInIdealState(tableNameWithType, segmentNames, () -> {
+ });
+ }
+
+ /// @param afterRead Runs after reading the IdealState and the segment ZK metadata, before removing it
+ @VisibleForTesting
+ Set<String> removeInProgressSegmentsZKMetadataNotInIdealState(String tableNameWithType,
+ Collection<String> segmentNames, Runnable afterRead) {
+ ZkClient zkClient = getOrBuildMultiWriteZkClient();
+ String idealStatePath = PropertyPathBuilder.idealState(_helixClusterName, tableNameWithType);
+ String propertyStoreRoot = PropertyPathBuilder.propertyStore(_helixClusterName);
+ for (int attempt = 1; ; attempt++) {
+ Stat idealStateStat = new Stat();
+ ZNRecord idealState = zkClient.readData(idealStatePath, idealStateStat);
Review Comment:
Minor/robustness: `zkClient.readData(path, stat)` returns `null` if the IdealState node is missing, which would NPE on `idealState.getMapFields()` below. Not reachable for a live realtime table mid-commit, so not blocking — but a null check that returns an empty set would make the helper safe to call defensively.
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManager.java:
##########
@@ -2278,6 +2283,80 @@ private ZkClient getOrBuildMultiWriteZkClient() {
}
}
+ /// Removes the ZK metadata of the given segments that are IN_PROGRESS and not in the IdealState of the table, i.e.
+ /// new consuming segments that were not added to the IdealState, and returns the segments whose ZK metadata was
+ /// removed.
+ ///
+ /// The removal is submitted in one ZK `multi()` transaction with version-checked deletes of the segment ZK metadata
+ /// read here, and a version-checked write of the unchanged IdealState:
+ /// - An IdealState update that read the IdealState before the transaction fails its version check. An updater that
+ /// reads the segment ZK metadata again when it retries, like the one of RealtimeSegmentValidationManager, then no
+ /// longer finds the removed segments.
+ /// - If the IdealState, or the ZK metadata of one of the segments, is updated after it was read here, the transaction
+ /// fails and is retried.
+ ///
+ /// The data versions cannot tell a segment ZK metadata deleted and recreated with the same name since it was read,
+ /// e.g. by an upload under the same name, from the one that was read. As the IdealState is outside the property
+ /// store, the transaction uses the multi-write ZkClient directly instead of [ZkMultiWriteBuilder].
+ public Set<String> removeInProgressSegmentsZKMetadataNotInIdealState(String tableNameWithType,
+ Collection<String> segmentNames) {
+ return removeInProgressSegmentsZKMetadataNotInIdealState(tableNameWithType, segmentNames, () -> {
+ });
+ }
+
+ /// @param afterRead Runs after reading the IdealState and the segment ZK metadata, before removing it
+ @VisibleForTesting
+ Set<String> removeInProgressSegmentsZKMetadataNotInIdealState(String tableNameWithType,
+ Collection<String> segmentNames, Runnable afterRead) {
+ ZkClient zkClient = getOrBuildMultiWriteZkClient();
+ String idealStatePath = PropertyPathBuilder.idealState(_helixClusterName, tableNameWithType);
+ String propertyStoreRoot = PropertyPathBuilder.propertyStore(_helixClusterName);
+ for (int attempt = 1; ; attempt++) {
+ Stat idealStateStat = new Stat();
+ ZNRecord idealState = zkClient.readData(idealStatePath, idealStateStat);
+ List<Op> segmentDeletes = new ArrayList<>();
+ Set<String> segmentsToRemove = new HashSet<>();
+ for (String segmentName : segmentNames) {
+ if (idealState.getMapFields().containsKey(segmentName)) {
+ continue;
+ }
+ String segmentPath =
+ propertyStoreRoot + ZKMetadataProvider.constructPropertyStorePathForSegment(tableNameWithType, segmentName);
+ Stat segmentStat = new Stat();
+ ZNRecord segmentRecord;
+ try {
+ segmentRecord = zkClient.readData(segmentPath, segmentStat);
+ } catch (ZkNoNodeException e) {
+ continue;
+ }
+ if (new SegmentZKMetadata(segmentRecord).getStatus() == CommonConstants.Segment.Realtime.Status.IN_PROGRESS) {
+ segmentDeletes.add(Op.delete(segmentPath, segmentStat.getVersion()));
+ segmentsToRemove.add(segmentName);
+ }
+ }
+ if (segmentsToRemove.isEmpty()) {
+ return segmentsToRemove;
+ }
+ afterRead.run();
+ List<Op> ops = new ArrayList<>(segmentDeletes.size() + 1);
+ ops.add(Op.setData(idealStatePath, zkClient.serialize(idealState, idealStatePath), idealStateStat.getVersion()));
+ ops.addAll(segmentDeletes);
+ try {
+ zkClient.multi(ops);
+ return segmentsToRemove;
+ } catch (ZkException e) {
Review Comment:
Minor: this catches `ZkException` and inspects `getCause()`, whereas the sibling `ZkMultiWriteBuilder.execute()` unwraps and rethrows the `KeeperException` subtype directly. Both work, but the two multi() call sites now handle the same failure differently. Worth a brief comment here (or reusing the builder's unwrap helper) so the next reader doesn't assume one of them is wrong.
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Remove the kept new segment when a resumed pauseless commit start completes without it [pinot]
Posted by "Jackie-Jiang (via GitHub)" <gi...@apache.org>.
Jackie-Jiang commented on code in PR #19995:
URL: https://github.com/apache/pinot/pull/19995#discussion_r4342857207
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManager.java:
##########
@@ -673,9 +673,13 @@ private void commitSegmentMetadataInternal(String realtimeTableName,
// Step-2: Create new segment metadata if needed
long startTimeNs2 = System.nanoTime();
+ // Another attempt of the same commit start might have created the new segment already. Find it before deciding
+ // whether the partition still needs a new segment, so that it can be removed when it is not used.
+ String newSegmentOfEarlierAttempt =
+ resumed ? findNewSegmentOfEarlierAttempt(realtimeTableName, new LLCSegmentName(committingSegmentName)) : null;
Review Comment:
Remaining pause-cleanup gap: more than one successor can be retained
The probe returns only the first matching name, but overlapping controllers can leave multiple successors with the same sequence number and different minute timestamps. For example, A marks the predecessor COMMITTING and stalls before step 2; new leader B resumes, finds no successor, creates N1, and fails its IdealState update. A then continues in a later minute with its original resumed=false, creates N2, and also fails. Both failure handlers retain their metadata while the predecessor is still CONSUMING.
A subsequent paused retry finds and removes only N1. N2 remains, so validation can still adopt it after the timeout and ignore the offset requested when the table resumes. This is a remaining variant of the original leak rather than a new regression in this PR.
Please add a two-retained-successor regression test and safely handle all unused candidates, preserving the protection against removing metadata for a segment adopted concurrently.
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManager.java:
##########
@@ -718,6 +722,15 @@ private void commitSegmentMetadataInternal(String realtimeTableName,
removeSegmentZKMetadataBestEffort(realtimeTableName, newConsumingSegmentName);
newConsumingSegmentName = null;
}
+ } else if (newSegmentOfEarlierAttempt != null
+ && !instanceStatesMapAfterStep3.containsKey(newSegmentOfEarlierAttempt)) {
+ // The committing segment is ONLINE now, so no attempt of the commit can add the new segment of the earlier
+ // attempt anymore. Remove it so that RealtimeSegmentValidationManager does not add it either.
+ LOGGER.info(
+ "Cleaning up segment ZK metadata for new segment {} of table {} created by an earlier attempt of the commit "
+ + "because it is not used. This can happen when table/topic consumption is paused, or the partition is "
+ + "no longer in the stream.", newSegmentOfEarlierAttempt, realtimeTableName);
+ removeSegmentZKMetadataBestEffort(realtimeTableName, newSegmentOfEarlierAttempt);
Review Comment:
[P1] Fence cleanup against validation adopting the retained segment
This deletion relies on the IdealState snapshot returned by step 3. A paused retry can mark the predecessor ONLINE and then stall before this removal. If the table resumes and the retained successor passes its completion timeout, validation can adopt that same successor as CONSUMING: its missing-previous-CONSUMING branch only logs an error, then calls updateInstanceStatesForNewConsumingSegment with a null predecessor. The stalled retry subsequently deletes the live successor's metadata using its stale snapshot.
The predecessor being ONLINE fences other commit attempts, but not this validation path. This interleaving uses the original retained segment across minutes; it does not require same-minute recreation. A retry near the completion deadline can make the required delay short.
Please coordinate deletion with validation's ability to adopt the segment (an ordinary IdealState re-read alone still leaves a race). Add a deterministic regression test that blocks cleanup before removal, lets resume/validation install the successor, and verifies that cleanup cannot remove its metadata.
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Remove the kept new segment when a resumed pauseless commit start completes without it [pinot]
Posted by "suvodeep-pyne (via GitHub)" <gi...@apache.org>.
suvodeep-pyne commented on code in PR #19995:
URL: https://github.com/apache/pinot/pull/19995#discussion_r4344943584
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManager.java:
##########
@@ -2278,6 +2283,80 @@ private ZkClient getOrBuildMultiWriteZkClient() {
}
}
+ /// Removes the ZK metadata of the given segments that are IN_PROGRESS and not in the IdealState of the table, i.e.
+ /// new consuming segments that were not added to the IdealState, and returns the segments whose ZK metadata was
+ /// removed.
+ ///
+ /// The removal is submitted in one ZK `multi()` transaction with version-checked deletes of the segment ZK metadata
+ /// read here, and a version-checked write of the unchanged IdealState:
+ /// - An IdealState update that read the IdealState before the transaction fails its version check. An updater that
+ /// reads the segment ZK metadata again when it retries, like the one of RealtimeSegmentValidationManager, then no
+ /// longer finds the removed segments.
+ /// - If the IdealState, or the ZK metadata of one of the segments, is updated after it was read here, the transaction
+ /// fails and is retried.
+ ///
+ /// The data versions cannot tell a segment ZK metadata deleted and recreated with the same name since it was read,
+ /// e.g. by an upload under the same name, from the one that was read. As the IdealState is outside the property
+ /// store, the transaction uses the multi-write ZkClient directly instead of [ZkMultiWriteBuilder].
+ public Set<String> removeInProgressSegmentsZKMetadataNotInIdealState(String tableNameWithType,
+ Collection<String> segmentNames) {
+ return removeInProgressSegmentsZKMetadataNotInIdealState(tableNameWithType, segmentNames, () -> {
+ });
+ }
+
+ /// @param afterRead Runs after reading the IdealState and the segment ZK metadata, before removing it
+ @VisibleForTesting
+ Set<String> removeInProgressSegmentsZKMetadataNotInIdealState(String tableNameWithType,
+ Collection<String> segmentNames, Runnable afterRead) {
+ ZkClient zkClient = getOrBuildMultiWriteZkClient();
+ String idealStatePath = PropertyPathBuilder.idealState(_helixClusterName, tableNameWithType);
+ String propertyStoreRoot = PropertyPathBuilder.propertyStore(_helixClusterName);
+ for (int attempt = 1; ; attempt++) {
+ Stat idealStateStat = new Stat();
+ ZNRecord idealState = zkClient.readData(idealStatePath, idealStateStat);
+ List<Op> segmentDeletes = new ArrayList<>();
+ Set<String> segmentsToRemove = new HashSet<>();
+ for (String segmentName : segmentNames) {
+ if (idealState.getMapFields().containsKey(segmentName)) {
+ continue;
+ }
+ String segmentPath =
+ propertyStoreRoot + ZKMetadataProvider.constructPropertyStorePathForSegment(tableNameWithType, segmentName);
+ Stat segmentStat = new Stat();
+ ZNRecord segmentRecord;
+ try {
+ segmentRecord = zkClient.readData(segmentPath, segmentStat);
+ } catch (ZkNoNodeException e) {
+ continue;
+ }
+ if (new SegmentZKMetadata(segmentRecord).getStatus() == CommonConstants.Segment.Realtime.Status.IN_PROGRESS) {
+ segmentDeletes.add(Op.delete(segmentPath, segmentStat.getVersion()));
+ segmentsToRemove.add(segmentName);
+ }
+ }
+ if (segmentsToRemove.isEmpty()) {
+ return segmentsToRemove;
+ }
+ afterRead.run();
+ List<Op> ops = new ArrayList<>(segmentDeletes.size() + 1);
+ ops.add(Op.setData(idealStatePath, zkClient.serialize(idealState, idealStatePath), idealStateStat.getVersion()));
+ ops.addAll(segmentDeletes);
+ try {
+ zkClient.multi(ops);
+ return segmentsToRemove;
+ } catch (ZkException e) {
Review Comment:
Added a comment in df8842da90. ZkMultiWriteBuilder can't take the IdealState path, since it only writes property-store paths, so this call site unwraps the same way inline.
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManager.java:
##########
@@ -2278,6 +2283,80 @@ private ZkClient getOrBuildMultiWriteZkClient() {
}
}
+ /// Removes the ZK metadata of the given segments that are IN_PROGRESS and not in the IdealState of the table, i.e.
+ /// new consuming segments that were not added to the IdealState, and returns the segments whose ZK metadata was
+ /// removed.
+ ///
+ /// The removal is submitted in one ZK `multi()` transaction with version-checked deletes of the segment ZK metadata
+ /// read here, and a version-checked write of the unchanged IdealState:
+ /// - An IdealState update that read the IdealState before the transaction fails its version check. An updater that
+ /// reads the segment ZK metadata again when it retries, like the one of RealtimeSegmentValidationManager, then no
+ /// longer finds the removed segments.
+ /// - If the IdealState, or the ZK metadata of one of the segments, is updated after it was read here, the transaction
+ /// fails and is retried.
+ ///
+ /// The data versions cannot tell a segment ZK metadata deleted and recreated with the same name since it was read,
+ /// e.g. by an upload under the same name, from the one that was read. As the IdealState is outside the property
+ /// store, the transaction uses the multi-write ZkClient directly instead of [ZkMultiWriteBuilder].
+ public Set<String> removeInProgressSegmentsZKMetadataNotInIdealState(String tableNameWithType,
+ Collection<String> segmentNames) {
+ return removeInProgressSegmentsZKMetadataNotInIdealState(tableNameWithType, segmentNames, () -> {
+ });
+ }
+
+ /// @param afterRead Runs after reading the IdealState and the segment ZK metadata, before removing it
+ @VisibleForTesting
+ Set<String> removeInProgressSegmentsZKMetadataNotInIdealState(String tableNameWithType,
+ Collection<String> segmentNames, Runnable afterRead) {
+ ZkClient zkClient = getOrBuildMultiWriteZkClient();
+ String idealStatePath = PropertyPathBuilder.idealState(_helixClusterName, tableNameWithType);
+ String propertyStoreRoot = PropertyPathBuilder.propertyStore(_helixClusterName);
+ for (int attempt = 1; ; attempt++) {
+ Stat idealStateStat = new Stat();
+ ZNRecord idealState = zkClient.readData(idealStatePath, idealStateStat);
Review Comment:
Done in df8842da90: a missing IdealState now returns an empty set without removing anything. It handles both a null result and the ZkNoNodeException that the ZkClient can throw for a missing node. The real-ZooKeeper test now also covers a deleted table.
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Remove the kept new segment when a resumed pauseless commit start completes without it [pinot]
Posted by "suvodeep-pyne (via GitHub)" <gi...@apache.org>.
suvodeep-pyne commented on PR #19995:
URL: https://github.com/apache/pinot/pull/19995#issuecomment-6354524743
Merged master into the branch (7abcc71b3f), which also brings in #19909's changes to the same controller files. On the rerun, every check passes except Pinot Integration Test Set 1, and that failure is unrelated to this change:
- `WindowFunnelTest.testFunnelCompleteCountGroupByQueries(boolean)[1]`: `expected [40] but found [36]`. It is a multi-stage funnel query on an offline table.
- Master hits the same failure in [run 39742128873](https://github.com/apache/pinot/actions/runs/39742128873) at 24c520be6e. It passes locally on this branch, so it looks intermittent.
@xiangfu0 could you re-run that job, or merge if you're comfortable with it as is?
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Remove the kept new segment when a resumed pauseless commit start completes without it [pinot]
Posted by "suvodeep-pyne (via GitHub)" <gi...@apache.org>.
suvodeep-pyne commented on PR #19995:
URL: https://github.com/apache/pinot/pull/19995#issuecomment-6336974291
The two failed checks on df8842da90 are unrelated to this change. Could a committer re-run them?
- Pinot Unit Test Set 1: `TarCompressionUtilsTest.testLz4ArchiveRoundTrip` in pinot-common (`arrays don't have the same size. expected [3163522] but found [3163523]`). It passed on the previous CI run of this PR, and passes locally.
- Pinot Unit Test Set 2: the build failed before running tests, with `FailedToAcquireLockException: Could not acquire shared lock for artifacts`.
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Remove the kept new segment when a resumed pauseless commit start completes without it [pinot]
Posted by "suvodeep-pyne (via GitHub)" <gi...@apache.org>.
suvodeep-pyne commented on PR #19995:
URL: https://github.com/apache/pinot/pull/19995#issuecomment-6371501999
All checks pass now on c5eac9b531, after merging master again. @xiangfu0 could you merge this when you get a chance? Thanks!
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Remove the kept new segment when a resumed pauseless commit start completes without it [pinot]
Posted by "Jackie-Jiang (via GitHub)" <gi...@apache.org>.
Jackie-Jiang merged PR #19995:
URL: https://github.com/apache/pinot/pull/19995
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org