You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pinot.apache.org by "Jackie-Jiang (via GitHub)" <gi...@apache.org> on 2026/07/23 21:45:25 UTC
[PR] Rebase and retry the IdealState update in place on a version conflict during tier-relocation rebalance [pinot]
Jackie-Jiang opened a new pull request, #19069:
URL: https://github.com/apache/pinot/pull/19069
## Summary
Follow-up to #19054. That PR reduces the *cost* of the version-checked `IdealState` update losing the compare-and-set during a tier-relocation rebalance of a strict realtime table (upsert/dedup); this PR reduces the cost of *recovering* from a lost compare-and-set.
Each batch of segment moves is applied with a version-checked `IdealState` update. On a continuously-ingesting table, consuming-segment commits bump the `IdealState` version between the read and the write, so the update loses the compare-and-set. Previously every lost compare-and-set fell back to the top of the convergence loop — another ExternalView-convergence wait (~hundreds of ms) plus a target recompute — so the rebalance could make very little progress while ingestion continued.
## Change
When the rebalance moves **only tier segments** (the base placements of the partitions are unchanged — the same condition #19054 uses to skip the target recompute), a concurrent write that does not touch the segments this batch moves cannot invalidate the batch. So on a `ZkBadVersionException`:
1. Re-read the `IdealState`.
2. If the concurrent change is **disjoint** from the segments this batch moves, rebase the batch onto the latest `IdealState` (reapply the batch's moves, preserving the concurrent changes) and retry the compare-and-set in place — no ExternalView wait, no `rebalanceTable`.
3. On overlap, re-read failure, or exhausting a bounded number of attempts (`MAX_IDEAL_STATE_UPDATE_REBASE_ATTEMPTS`), fall back to the convergence loop as before.
On a successful rebase, the target assignment is refreshed from the adopted `IdealState` so the `currentAssignment.equals(targetAssignment)` convergence check stays well-defined when a rebase pulls in concurrently added (e.g. new consuming) or removed (e.g. retention) segments.
Net effect: disjoint version churn is absorbed in place, so a tier relocation makes progress against a steady stream of consuming-segment commits instead of repeatedly restarting.
## Notes
- The rebase is gated on the same "moves only tier segments" check as #19054, so when the rebalance moves base placements it conservatively keeps the existing fall-back behavior.
- Reviewers: the observer stats handling on a successful rebase (`onRollback` is skipped since the batch did land) is the subtlest part and the best target for a unit test that injects a version bump between the read and the compare-and-set.
--
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] Rebase and retry the IdealState update in place on a version conflict during tier-relocation rebalance [pinot]
Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #19069:
URL: https://github.com/apache/pinot/pull/19069#issuecomment-5064231348
## [Codecov](https://app.codecov.io/gh/apache/pinot/pull/19069?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 56.84%. Comparing base ([`867b53b`](https://app.codecov.io/gh/apache/pinot/commit/867b53b7489d476407a09ca5f40b12a42693e58c?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)) to head ([`ddcba5e`](https://app.codecov.io/gh/apache/pinot/commit/ddcba5e98902524be8203a9ac7fd14cc7dabc7ee?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)).
:warning: Report is 1 commits behind head on master.
> :exclamation: There is a different number of reports uploaded between BASE (867b53b) and HEAD (ddcba5e). Click for more details.
>
> <details><summary>HEAD has 16 uploads less than BASE</summary>
>
>| Flag | BASE (867b53b) | HEAD (ddcba5e) |
>|------|------|------|
>|unittests|2|1|
>|java-25|5|1|
>|temurin|5|1|
>|unittests2|1|0|
>|integration|3|0|
>|integration1|1|0|
>|integration2|1|0|
>|custom-integration1|1|0|
></details>
<details><summary>Additional details and impacted files</summary>
```diff
@@ Coverage Diff @@
## master #19069 +/- ##
============================================
- Coverage 65.43% 56.84% -8.60%
+ Complexity 1421 1 -1420
============================================
Files 3425 2628 -797
Lines 216169 156488 -59681
Branches 34236 25579 -8657
============================================
- Hits 141460 88949 -52511
+ Misses 63327 59925 -3402
+ Partials 11382 7614 -3768
```
| [Flag](https://app.codecov.io/gh/apache/pinot/pull/19069/flags?src=pr&el=flags&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [custom-integration1](https://app.codecov.io/gh/apache/pinot/pull/19069/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
| [integration](https://app.codecov.io/gh/apache/pinot/pull/19069/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
| [integration1](https://app.codecov.io/gh/apache/pinot/pull/19069/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
| [integration2](https://app.codecov.io/gh/apache/pinot/pull/19069/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/19069/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `56.84% <ø> (-8.60%)` | :arrow_down: |
| [temurin](https://app.codecov.io/gh/apache/pinot/pull/19069/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `56.84% <ø> (-8.60%)` | :arrow_down: |
| [unittests](https://app.codecov.io/gh/apache/pinot/pull/19069/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `56.84% <ø> (-8.60%)` | :arrow_down: |
| [unittests1](https://app.codecov.io/gh/apache/pinot/pull/19069/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `56.84% <ø> (-0.02%)` | :arrow_down: |
| [unittests2](https://app.codecov.io/gh/apache/pinot/pull/19069/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/19069?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] Rebase and retry the IdealState update in place on a version conflict during tier-relocation rebalance [pinot]
Posted by "J-HowHuang (via GitHub)" <gi...@apache.org>.
J-HowHuang commented on code in PR #19069:
URL: https://github.com/apache/pinot/pull/19069#discussion_r3669603170
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/rebalance/TableRebalancer.java:
##########
@@ -812,27 +816,105 @@ private RebalanceResult doRebalance(TableConfig tableConfig, RebalanceConfig reb
idealState.setNumPartitions(nextAssignment.size());
idealState.setReplicas(Integer.toString(nextAssignment.values().iterator().next().size()));
- // Check version and update IdealState
- try {
- Preconditions.checkState(_helixDataAccessor.getBaseDataAccessor()
- .set(idealStatePropertyKey.getPath(), idealStateRecord, expectedVersion, AccessOption.PERSISTENT),
- "Failed to update IdealState");
+ // Segments this batch changes relative to the current assignment it was computed against. Captured before the
+ // compare-and-set so that, on a version conflict, we can tell whether a concurrent write touched any segment
+ // this batch moves.
+ List<String> batchMovedSegments = SegmentAssignmentUtils.getSegmentsToMove(currentAssignment, nextAssignment);
+
+ // Check version and update the IdealState. If the compare-and-set fails only because a concurrent write bumped
+ // the version without touching the segments this batch moves (e.g. consuming segment commits on a continuously
+ // ingesting table), rebase this batch onto the latest IdealState and retry the compare-and-set in place, without
+ // waiting for the ExternalView to converge again (this batch never landed, so there is nothing new to wait for)
+ // or recomputing the full target assignment. This keeps the rebalance from live-locking against a steady stream
+ // of version bumps. Only attempted when this rebalance moves only tier segments: the base placements are then
+ // unchanged, so a segment added concurrently keeps its correct placement and can be carried over as-is.
+ boolean rebasable = isMovingOnlyTierSegments(segmentsToMove, providedTierToSegmentsMap);
Review Comment:
Should this be
```suggestion
boolean rebasable = !isStrictRealtimeSegmentAssignment || isMovingOnlyTierSegments(batchMovedSegments, providedTierToSegmentsMap);
```
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/rebalance/TableRebalancer.java:
##########
@@ -812,27 +816,105 @@ private RebalanceResult doRebalance(TableConfig tableConfig, RebalanceConfig reb
idealState.setNumPartitions(nextAssignment.size());
idealState.setReplicas(Integer.toString(nextAssignment.values().iterator().next().size()));
- // Check version and update IdealState
- try {
- Preconditions.checkState(_helixDataAccessor.getBaseDataAccessor()
- .set(idealStatePropertyKey.getPath(), idealStateRecord, expectedVersion, AccessOption.PERSISTENT),
- "Failed to update IdealState");
+ // Segments this batch changes relative to the current assignment it was computed against. Captured before the
+ // compare-and-set so that, on a version conflict, we can tell whether a concurrent write touched any segment
+ // this batch moves.
+ List<String> batchMovedSegments = SegmentAssignmentUtils.getSegmentsToMove(currentAssignment, nextAssignment);
+
+ // Check version and update the IdealState. If the compare-and-set fails only because a concurrent write bumped
+ // the version without touching the segments this batch moves (e.g. consuming segment commits on a continuously
+ // ingesting table), rebase this batch onto the latest IdealState and retry the compare-and-set in place, without
+ // waiting for the ExternalView to converge again (this batch never landed, so there is nothing new to wait for)
+ // or recomputing the full target assignment. This keeps the rebalance from live-locking against a steady stream
+ // of version bumps. Only attempted when this rebalance moves only tier segments: the base placements are then
+ // unchanged, so a segment added concurrently keeps its correct placement and can be carried over as-is.
+ boolean rebasable = isMovingOnlyTierSegments(segmentsToMove, providedTierToSegmentsMap);
+ boolean updated = false;
+ int rebaseAttempts = 0;
+ while (true) {
+ try {
+ Preconditions.checkState(_helixDataAccessor.getBaseDataAccessor()
+ .set(idealStatePropertyKey.getPath(), idealStateRecord, expectedVersion, AccessOption.PERSISTENT),
+ "Failed to update IdealState");
+ updated = true;
+ break;
+ } catch (ZkBadVersionException e) {
+ if (!rebasable || rebaseAttempts >= MAX_IDEAL_STATE_UPDATE_REBASE_ATTEMPTS) {
+ break;
+ }
+ IdealState latestIdealState;
+ try {
+ latestIdealState = _helixDataAccessor.getProperty(idealStatePropertyKey);
+ } catch (Exception re) {
+ tableRebalanceLogger.warn("Failed to re-read IdealState for rebasing after a version conflict", re);
+ break;
+ }
+ if (latestIdealState == null) {
+ break;
+ }
+ Map<String, Map<String, String>> latestAssignment = latestIdealState.getRecord().getMapFields();
+ boolean concurrentChangeTouchesBatch = false;
+ for (String segment : batchMovedSegments) {
+ if (!Objects.equals(currentAssignment.get(segment), latestAssignment.get(segment))) {
+ concurrentChangeTouchesBatch = true;
+ break;
+ }
+ }
+ if (concurrentChangeTouchesBatch) {
+ break;
+ }
+ // Rebase the batch onto the latest IdealState, preserving concurrent changes to the other segments.
+ Map<String, Map<String, String>> rebasedAssignment = new TreeMap<>(latestAssignment);
+ for (String segment : batchMovedSegments) {
+ rebasedAssignment.put(segment, nextAssignment.get(segment));
+ }
+ nextAssignment = rebasedAssignment;
+ idealState = latestIdealState;
+ idealStateRecord = latestIdealState.getRecord();
+ idealStateRecord.setMapFields(nextAssignment);
+ idealState.setNumPartitions(nextAssignment.size());
+ idealState.setReplicas(Integer.toString(nextAssignment.values().iterator().next().size()));
+ expectedVersion = idealStateRecord.getVersion();
+ rebaseAttempts++;
+ tableRebalanceLogger.info("Rebasing IdealState update onto version {} after a concurrent change that does "
+ + "not affect the segments being moved (attempt {})", expectedVersion, rebaseAttempts);
+ } catch (Exception e) {
+ onReturnFailure("Caught exception while updating IdealState, aborting the rebalance", e,
+ tableRebalanceLogger);
+ return new RebalanceResult(rebalanceJobId, RebalanceResult.Status.FAILED,
+ "Caught exception while updating IdealState: " + e, instancePartitionsMap, tierToInstancePartitionsMap,
+ targetAssignment, preChecksResult, summaryResult);
+ }
+ }
+
+ if (updated) {
expectedVersion++;
currentAssignment = nextAssignment;
+ if (rebaseAttempts > 0) {
+ // A rebase adopts the latest IdealState, which may have segments added (e.g. new consuming segments) or
+ // removed (e.g. by retention) concurrently that the target assignment, computed earlier, does not match.
+ // Since a rebase is only done when the rebalance moves only tier segments, added segments keep their current
+ // placement, so rebuild the target from the adopted current assignment while preserving the target only for
+ // the segments still moving that still exist. This keeps the convergence check
+ // (currentAssignment.equals(targetAssignment)) well-defined and its key set aligned with the current
+ // assignment.
+ Map<String, Map<String, String>> refreshedTarget = new TreeMap<>(currentAssignment);
+ for (String segment : SegmentAssignmentUtils.getSegmentsToMove(currentAssignment, targetAssignment)) {
+ if (currentAssignment.containsKey(segment)) {
+ refreshedTarget.put(segment, targetAssignment.get(segment));
+ }
+ }
+ targetAssignment = refreshedTarget;
Review Comment:
we may also reuse the `rebaseTargetAssignment` method if we have one here.
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/rebalance/TableRebalancer.java:
##########
@@ -812,27 +816,105 @@ private RebalanceResult doRebalance(TableConfig tableConfig, RebalanceConfig reb
idealState.setNumPartitions(nextAssignment.size());
idealState.setReplicas(Integer.toString(nextAssignment.values().iterator().next().size()));
- // Check version and update IdealState
- try {
- Preconditions.checkState(_helixDataAccessor.getBaseDataAccessor()
- .set(idealStatePropertyKey.getPath(), idealStateRecord, expectedVersion, AccessOption.PERSISTENT),
- "Failed to update IdealState");
+ // Segments this batch changes relative to the current assignment it was computed against. Captured before the
+ // compare-and-set so that, on a version conflict, we can tell whether a concurrent write touched any segment
+ // this batch moves.
+ List<String> batchMovedSegments = SegmentAssignmentUtils.getSegmentsToMove(currentAssignment, nextAssignment);
+
+ // Check version and update the IdealState. If the compare-and-set fails only because a concurrent write bumped
+ // the version without touching the segments this batch moves (e.g. consuming segment commits on a continuously
+ // ingesting table), rebase this batch onto the latest IdealState and retry the compare-and-set in place, without
+ // waiting for the ExternalView to converge again (this batch never landed, so there is nothing new to wait for)
+ // or recomputing the full target assignment. This keeps the rebalance from live-locking against a steady stream
+ // of version bumps. Only attempted when this rebalance moves only tier segments: the base placements are then
+ // unchanged, so a segment added concurrently keeps its correct placement and can be carried over as-is.
+ boolean rebasable = isMovingOnlyTierSegments(segmentsToMove, providedTierToSegmentsMap);
+ boolean updated = false;
+ int rebaseAttempts = 0;
+ while (true) {
+ try {
+ Preconditions.checkState(_helixDataAccessor.getBaseDataAccessor()
+ .set(idealStatePropertyKey.getPath(), idealStateRecord, expectedVersion, AccessOption.PERSISTENT),
+ "Failed to update IdealState");
+ updated = true;
+ break;
+ } catch (ZkBadVersionException e) {
+ if (!rebasable || rebaseAttempts >= MAX_IDEAL_STATE_UPDATE_REBASE_ATTEMPTS) {
+ break;
+ }
+ IdealState latestIdealState;
+ try {
+ latestIdealState = _helixDataAccessor.getProperty(idealStatePropertyKey);
+ } catch (Exception re) {
+ tableRebalanceLogger.warn("Failed to re-read IdealState for rebasing after a version conflict", re);
+ break;
+ }
+ if (latestIdealState == null) {
+ break;
+ }
+ Map<String, Map<String, String>> latestAssignment = latestIdealState.getRecord().getMapFields();
+ boolean concurrentChangeTouchesBatch = false;
+ for (String segment : batchMovedSegments) {
+ if (!Objects.equals(currentAssignment.get(segment), latestAssignment.get(segment))) {
+ concurrentChangeTouchesBatch = true;
+ break;
+ }
+ }
+ if (concurrentChangeTouchesBatch) {
+ break;
+ }
+ // Rebase the batch onto the latest IdealState, preserving concurrent changes to the other segments.
+ Map<String, Map<String, String>> rebasedAssignment = new TreeMap<>(latestAssignment);
+ for (String segment : batchMovedSegments) {
+ rebasedAssignment.put(segment, nextAssignment.get(segment));
+ }
+ nextAssignment = rebasedAssignment;
Review Comment:
nit: Can we have a method `rebaseTargetAssignment(originalTargetAssignment, originalCurrentAssignment, newCurrentAssingment)` that resolves `rebasable` and `rebasedAssignment`?
Then we can use the same method here: https://github.com/apache/pinot/pull/19069/changes#diff-d4962fcb9ad5591bb650b16bec8858ae3d9aa357aadc5cf2940a0524216898aaR651-R720, it's basically doing the same thing only that it rebases target assignment and here we rebase next assignment
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/rebalance/TableRebalancer.java:
##########
@@ -812,27 +816,105 @@ private RebalanceResult doRebalance(TableConfig tableConfig, RebalanceConfig reb
idealState.setNumPartitions(nextAssignment.size());
idealState.setReplicas(Integer.toString(nextAssignment.values().iterator().next().size()));
- // Check version and update IdealState
- try {
- Preconditions.checkState(_helixDataAccessor.getBaseDataAccessor()
- .set(idealStatePropertyKey.getPath(), idealStateRecord, expectedVersion, AccessOption.PERSISTENT),
- "Failed to update IdealState");
+ // Segments this batch changes relative to the current assignment it was computed against. Captured before the
+ // compare-and-set so that, on a version conflict, we can tell whether a concurrent write touched any segment
+ // this batch moves.
+ List<String> batchMovedSegments = SegmentAssignmentUtils.getSegmentsToMove(currentAssignment, nextAssignment);
+
+ // Check version and update the IdealState. If the compare-and-set fails only because a concurrent write bumped
+ // the version without touching the segments this batch moves (e.g. consuming segment commits on a continuously
+ // ingesting table), rebase this batch onto the latest IdealState and retry the compare-and-set in place, without
+ // waiting for the ExternalView to converge again (this batch never landed, so there is nothing new to wait for)
+ // or recomputing the full target assignment. This keeps the rebalance from live-locking against a steady stream
+ // of version bumps. Only attempted when this rebalance moves only tier segments: the base placements are then
+ // unchanged, so a segment added concurrently keeps its correct placement and can be carried over as-is.
+ boolean rebasable = isMovingOnlyTierSegments(segmentsToMove, providedTierToSegmentsMap);
+ boolean updated = false;
+ int rebaseAttempts = 0;
+ while (true) {
Review Comment:
I think in this case it would go to https://github.com/apache/pinot/pull/19069/changes#diff-d4962fcb9ad5591bb650b16bec8858ae3d9aa357aadc5cf2940a0524216898aaR651, so it should be fine? Please confirm this as well.
--
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] Rebase and retry the IdealState update in place on a version conflict during tier-relocation rebalance [pinot]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #19069:
URL: https://github.com/apache/pinot/pull/19069#discussion_r3669208990
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/rebalance/TableRebalancer.java:
##########
@@ -812,27 +816,105 @@ private RebalanceResult doRebalance(TableConfig tableConfig, RebalanceConfig reb
idealState.setNumPartitions(nextAssignment.size());
idealState.setReplicas(Integer.toString(nextAssignment.values().iterator().next().size()));
- // Check version and update IdealState
- try {
- Preconditions.checkState(_helixDataAccessor.getBaseDataAccessor()
- .set(idealStatePropertyKey.getPath(), idealStateRecord, expectedVersion, AccessOption.PERSISTENT),
- "Failed to update IdealState");
+ // Segments this batch changes relative to the current assignment it was computed against. Captured before the
+ // compare-and-set so that, on a version conflict, we can tell whether a concurrent write touched any segment
+ // this batch moves.
+ List<String> batchMovedSegments = SegmentAssignmentUtils.getSegmentsToMove(currentAssignment, nextAssignment);
+
+ // Check version and update the IdealState. If the compare-and-set fails only because a concurrent write bumped
+ // the version without touching the segments this batch moves (e.g. consuming segment commits on a continuously
+ // ingesting table), rebase this batch onto the latest IdealState and retry the compare-and-set in place, without
+ // waiting for the ExternalView to converge again (this batch never landed, so there is nothing new to wait for)
+ // or recomputing the full target assignment. This keeps the rebalance from live-locking against a steady stream
+ // of version bumps. Only attempted when this rebalance moves only tier segments: the base placements are then
+ // unchanged, so a segment added concurrently keeps its correct placement and can be carried over as-is.
+ boolean rebasable = isMovingOnlyTierSegments(segmentsToMove, providedTierToSegmentsMap);
+ boolean updated = false;
+ int rebaseAttempts = 0;
+ while (true) {
Review Comment:
`expectedVersion` is later mutated during rebase attempts (set to the latest IdealState version) even if the update ultimately fails. That breaks the invariant that `expectedVersion` matches the version of `currentAssignment` (which is only updated on a successful write or when the version-mismatch block runs). If rebasing hits the max attempts and falls back, the next loop iteration can incorrectly skip the version-mismatch path and continue with a stale `currentAssignment`/`targetAssignment`. Capture the original version before attempting rebases so it can be restored on failure.
This issue also appears on line 914 of the same file.
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/rebalance/TableRebalancer.java:
##########
@@ -812,27 +816,105 @@ private RebalanceResult doRebalance(TableConfig tableConfig, RebalanceConfig reb
idealState.setNumPartitions(nextAssignment.size());
idealState.setReplicas(Integer.toString(nextAssignment.values().iterator().next().size()));
- // Check version and update IdealState
- try {
- Preconditions.checkState(_helixDataAccessor.getBaseDataAccessor()
- .set(idealStatePropertyKey.getPath(), idealStateRecord, expectedVersion, AccessOption.PERSISTENT),
- "Failed to update IdealState");
+ // Segments this batch changes relative to the current assignment it was computed against. Captured before the
+ // compare-and-set so that, on a version conflict, we can tell whether a concurrent write touched any segment
+ // this batch moves.
+ List<String> batchMovedSegments = SegmentAssignmentUtils.getSegmentsToMove(currentAssignment, nextAssignment);
+
+ // Check version and update the IdealState. If the compare-and-set fails only because a concurrent write bumped
+ // the version without touching the segments this batch moves (e.g. consuming segment commits on a continuously
+ // ingesting table), rebase this batch onto the latest IdealState and retry the compare-and-set in place, without
+ // waiting for the ExternalView to converge again (this batch never landed, so there is nothing new to wait for)
+ // or recomputing the full target assignment. This keeps the rebalance from live-locking against a steady stream
+ // of version bumps. Only attempted when this rebalance moves only tier segments: the base placements are then
+ // unchanged, so a segment added concurrently keeps its correct placement and can be carried over as-is.
Review Comment:
This change adds fairly subtle state-machine behavior (rebasing an IdealState batch update on `ZkBadVersionException`, plus refreshing `targetAssignment` after adopting a newer IdealState). There isn’t a unit test covering the version-conflict/rebase path today, even though `pinot-controller` already has `TableRebalancerTest`. Adding a regression test that forces a version bump between the initial read and the compare-and-set (and asserts the rebased update preserves disjoint concurrent changes and keeps observer stats consistent) would help prevent future regressions.
--
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