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 01:07:41 UTC
[PR] Skip full target recompute on version change for strict realtime tier relocation [pinot]
Jackie-Jiang opened a new pull request, #19054:
URL: https://github.com/apache/pinot/pull/19054
## Summary
During a tier-relocation rebalance of a strict realtime table (upsert/dedup), `TableRebalancer`'s convergence loop reads the `IdealState` at version `V`, recomputes the target assignment, then issues a version-checked update expecting `V`. 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 and retries.
For strict realtime segment assignment, the recompute on a version change was **unconditionally** a full `rebalanceTable`, which reads segment ZK metadata per partition (`MultiTierStrictRealtimeSegmentAssignment.getExistingAssignment`). That made every retry expensive and widened the window in which the version can change, so a tier relocation could make very little progress while ingestion continued.
The full recompute is only necessary when the rebalance actually **moves consuming segments** (e.g. rebalancing the consuming tier). When it only moves completed segments (a tier relocation), a consuming segment added to the `IdealState` mid-rebalance keeps its placement in both the current and target assignment, so the previously computed target can be reused via the existing cheap path — the same one non-strict assignment already uses when no moving segment changed state.
## Change
- Gate the forced full recompute on `isStrictRealtimeSegmentAssignment && hasMovingConsumingSegments(...)` instead of `instanceof BaseStrictRealtimeSegmentAssignment`. For a tier relocation (no consuming segments moving), the loop now takes the cheap path on a version change, narrowing the update window and letting the rebalance make progress under sustained ingestion.
- The segment `tier` ZK metadata is persisted up front by `TableRebalanceManager.updateTargetTier` before the convergence loop, so it does not change per iteration — reusing the target during the loop is safe.
- Simplify `getMovingConsumingSegments` to a single pass that reads the first entry of each segment's (uniform) instance-state map, and add a short-circuiting `hasMovingConsumingSegments` for the existence check the gate uses.
- Add `testHasMovingConsumingSegments` and update `testGetMovingConsumingSegments` to realistic uniform-state inputs.
## Notes
- This is the first of two changes. A follow-up will add rebase-on-conflict to the versioned `IdealState` update, so a lost compare-and-set can retry against the newest version without another ExternalView wait; this PR only reduces the cost and likelihood of the conflict.
- Trade-off: a tier relocation now skips the mid-loop instance-partition recompute, so an instance-config change *during* a relocation is picked up on the next rebalance run rather than mid-flight.
--
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] Skip full target recompute on IdealState version change for strict realtime rebalance without instance reassignment [pinot]
Posted by "Jackie-Jiang (via GitHub)" <gi...@apache.org>.
Jackie-Jiang commented on PR #19054:
URL: https://github.com/apache/pinot/pull/19054#issuecomment-5055372034
@J-HowHuang thanks — your paused-table scenario was spot on and showed the consuming-segment-based gate was insufficient: segments added mid-rebalance by a minion task would not have been re-collocated. The PR has been reworked to gate on instance reassignment instead.
A newly added segment (new consuming segment or uploaded segment) is placed from the current instance partitions, collocated with its partition. So when the rebalance is not reassigning instances (`reassignInstances`/`bootstrap` both false — e.g. the `SegmentRelocator` flow), the instance partitions are read-only for the job and any segment added mid-rebalance is placed consistently with the target, making the cheap path safe. With instance reassignment enabled, the conservative full recompute is kept, matching prior behavior. In your scenario the config-change rebalance either reassigns instances (full recompute, re-collocated) or does not (added segments are placed from the same fixed instance partitions the target uses), so both cases are handled.
@xiangfu0 the reworked PR no longer touches `getMovingConsumingSegments()`, so the first-replica inference and the removed test case are both back to the original. PR description is updated to match.
--
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] Skip full target recompute on version change for strict realtime tier relocation [pinot]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #19054:
URL: https://github.com/apache/pinot/pull/19054#discussion_r3634946288
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/rebalance/TableRebalancer.java:
##########
@@ -2233,20 +2248,40 @@ static Set<String> getMovingConsumingSegments(Map<String, Map<String, String>> c
Map<String, Map<String, String>> targetAssignment) {
Set<String> movingConsumingSegments = new HashSet<>();
for (Map.Entry<String, Map<String, String>> entry : currentAssignment.entrySet()) {
- String segmentName = entry.getKey();
- Map<String, String> currentInstanceStateMap = entry.getValue();
- Map<String, String> targetInstanceStateMap = targetAssignment.get(segmentName);
- if (targetInstanceStateMap != null && targetInstanceStateMap.values().stream()
- .noneMatch(state -> state.equals(SegmentStateModel.ONLINE)) && targetInstanceStateMap.values().stream()
- .anyMatch(state -> state.equals(SegmentStateModel.CONSUMING))) {
- if (!currentInstanceStateMap.keySet().equals(targetInstanceStateMap.keySet())) {
- movingConsumingSegments.add(segmentName);
- }
+ if (isMovingConsumingSegment(entry.getValue(), targetAssignment.get(entry.getKey()))) {
+ movingConsumingSegments.add(entry.getKey());
}
}
return movingConsumingSegments;
}
+ /// Returns whether any consuming segment moves between the current and target assignment, short-circuiting on the
+ /// first such segment. Cheaper than `!getMovingConsumingSegments(...).isEmpty()` when only existence is needed
+ /// because it does not materialize the full set.
+ @VisibleForTesting
+ static boolean hasMovingConsumingSegments(Map<String, Map<String, String>> currentAssignment,
+ Map<String, Map<String, String>> targetAssignment) {
+ for (Map.Entry<String, Map<String, String>> entry : currentAssignment.entrySet()) {
+ if (isMovingConsumingSegment(entry.getValue(), targetAssignment.get(entry.getKey()))) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ /// Returns whether the segment is a consuming segment being moved: its target state is `CONSUMING` and its assigned
+ /// instances differ between the current and target assignment. A segment's target instance state map is uniform
+ /// (every replica is `ONLINE`, `CONSUMING`, or `OFFLINE`) because `RealtimeSegmentAssignment` assigns a single state
+ /// to all instances of a moved segment and keeps `OFFLINE` segments unchanged, so the segment's state can be read
+ /// from the first entry. The map is non-null and non-empty: callers iterate the current assignment and look up a
Review Comment:
The new method comment says the target instance-state map is uniform and that the segment state can be read from the first entry. That’s not true for intermediate `nextAssignment` maps built by `getNextSingleSegmentAssignment()` (which can mix states), and it would also become inaccurate if `isMovingConsumingSegment()` is fixed to scan values.
This issue also appears on line 2279 of the same file.
##########
pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/rebalance/TableRebalancerClusterStatelessTest.java:
##########
@@ -2392,67 +2392,99 @@ public void testRebalanceConsumingSegmentSummaryFailure()
@Test
public void testGetMovingConsumingSegments() {
- // Setup: segment assignments with consuming segments moving
+ // A segment's instance state map is uniform: every replica shares one state (ONLINE, CONSUMING, or OFFLINE), so
+ // each segment below uses a single state across all its instances.
Review Comment:
The test now states instance-state maps are always uniform, but `TableRebalancer.getNextSingleSegmentAssignment()` can produce a mixed state map (e.g., keeping an ONLINE replica to satisfy min-available-replicas while others are CONSUMING). This test should cover that mixed-state shape to avoid regressions like relying on `iterator().next()` to infer the segment state.
This issue also appears on line 2430 of the same file.
--
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] Skip full target recompute on IdealState version change for strict realtime rebalance without instance reassignment [pinot]
Posted by "Jackie-Jiang (via GitHub)" <gi...@apache.org>.
Jackie-Jiang commented on code in PR #19054:
URL: https://github.com/apache/pinot/pull/19054#discussion_r3636119013
##########
pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/rebalance/TableRebalancerClusterStatelessTest.java:
##########
@@ -2392,67 +2392,99 @@ public void testRebalanceConsumingSegmentSummaryFailure()
@Test
public void testGetMovingConsumingSegments() {
- // Setup: segment assignments with consuming segments moving
+ // A segment's instance state map is uniform: every replica shares one state (ONLINE, CONSUMING, or OFFLINE), so
+ // each segment below uses a single state across all its instances.
Review Comment:
The PR has been reworked: `getMovingConsumingSegments()` and its test are no longer modified, so the uniformity assumption this comment flagged is gone. The recompute gate is now based on `reassignInstances`/`bootstrap` instead.
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/rebalance/TableRebalancer.java:
##########
@@ -2233,20 +2248,40 @@ static Set<String> getMovingConsumingSegments(Map<String, Map<String, String>> c
Map<String, Map<String, String>> targetAssignment) {
Set<String> movingConsumingSegments = new HashSet<>();
for (Map.Entry<String, Map<String, String>> entry : currentAssignment.entrySet()) {
- String segmentName = entry.getKey();
- Map<String, String> currentInstanceStateMap = entry.getValue();
- Map<String, String> targetInstanceStateMap = targetAssignment.get(segmentName);
- if (targetInstanceStateMap != null && targetInstanceStateMap.values().stream()
- .noneMatch(state -> state.equals(SegmentStateModel.ONLINE)) && targetInstanceStateMap.values().stream()
- .anyMatch(state -> state.equals(SegmentStateModel.CONSUMING))) {
- if (!currentInstanceStateMap.keySet().equals(targetInstanceStateMap.keySet())) {
- movingConsumingSegments.add(segmentName);
- }
+ if (isMovingConsumingSegment(entry.getValue(), targetAssignment.get(entry.getKey()))) {
+ movingConsumingSegments.add(entry.getKey());
}
}
return movingConsumingSegments;
}
+ /// Returns whether any consuming segment moves between the current and target assignment, short-circuiting on the
+ /// first such segment. Cheaper than `!getMovingConsumingSegments(...).isEmpty()` when only existence is needed
+ /// because it does not materialize the full set.
+ @VisibleForTesting
+ static boolean hasMovingConsumingSegments(Map<String, Map<String, String>> currentAssignment,
+ Map<String, Map<String, String>> targetAssignment) {
+ for (Map.Entry<String, Map<String, String>> entry : currentAssignment.entrySet()) {
+ if (isMovingConsumingSegment(entry.getValue(), targetAssignment.get(entry.getKey()))) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ /// Returns whether the segment is a consuming segment being moved: its target state is `CONSUMING` and its assigned
+ /// instances differ between the current and target assignment. A segment's target instance state map is uniform
+ /// (every replica is `ONLINE`, `CONSUMING`, or `OFFLINE`) because `RealtimeSegmentAssignment` assigns a single state
+ /// to all instances of a moved segment and keeps `OFFLINE` segments unchanged, so the segment's state can be read
+ /// from the first entry. The map is non-null and non-empty: callers iterate the current assignment and look up a
Review Comment:
The PR has been reworked: `getMovingConsumingSegments()` and its test are no longer modified, so the uniformity assumption this comment flagged is gone. The recompute gate is now based on `reassignInstances`/`bootstrap` instead.
--
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] Skip full target recompute on version change for strict realtime tier relocation [pinot]
Posted by "xiangfu0 (via GitHub)" <gi...@apache.org>.
xiangfu0 commented on code in PR #19054:
URL: https://github.com/apache/pinot/pull/19054#discussion_r3635509479
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/rebalance/TableRebalancer.java:
##########
@@ -2233,20 +2248,40 @@ static Set<String> getMovingConsumingSegments(Map<String, Map<String, String>> c
Map<String, Map<String, String>> targetAssignment) {
Set<String> movingConsumingSegments = new HashSet<>();
for (Map.Entry<String, Map<String, String>> entry : currentAssignment.entrySet()) {
- String segmentName = entry.getKey();
- Map<String, String> currentInstanceStateMap = entry.getValue();
- Map<String, String> targetInstanceStateMap = targetAssignment.get(segmentName);
- if (targetInstanceStateMap != null && targetInstanceStateMap.values().stream()
- .noneMatch(state -> state.equals(SegmentStateModel.ONLINE)) && targetInstanceStateMap.values().stream()
- .anyMatch(state -> state.equals(SegmentStateModel.CONSUMING))) {
- if (!currentInstanceStateMap.keySet().equals(targetInstanceStateMap.keySet())) {
- movingConsumingSegments.add(segmentName);
- }
+ if (isMovingConsumingSegment(entry.getValue(), targetAssignment.get(entry.getKey()))) {
+ movingConsumingSegments.add(entry.getKey());
}
}
return movingConsumingSegments;
}
+ /// Returns whether any consuming segment moves between the current and target assignment, short-circuiting on the
+ /// first such segment. Cheaper than `!getMovingConsumingSegments(...).isEmpty()` when only existence is needed
+ /// because it does not materialize the full set.
+ @VisibleForTesting
+ static boolean hasMovingConsumingSegments(Map<String, Map<String, String>> currentAssignment,
+ Map<String, Map<String, String>> targetAssignment) {
+ for (Map.Entry<String, Map<String, String>> entry : currentAssignment.entrySet()) {
+ if (isMovingConsumingSegment(entry.getValue(), targetAssignment.get(entry.getKey()))) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ /// Returns whether the segment is a consuming segment being moved: its target state is `CONSUMING` and its assigned
+ /// instances differ between the current and target assignment. A segment's target instance state map is uniform
+ /// (every replica is `ONLINE`, `CONSUMING`, or `OFFLINE`) because `RealtimeSegmentAssignment` assigns a single state
+ /// to all instances of a moved segment and keeps `OFFLINE` segments unchanged, so the segment's state can be read
+ /// from the first entry. The map is non-null and non-empty: callers iterate the current assignment and look up a
+ /// target derived from it (via `rebalanceTable` / `getNextAssignment`), which covers every current segment, and
+ /// every segment has at least one replica.
+ private static boolean isMovingConsumingSegment(Map<String, String> currentInstanceStateMap,
+ Map<String, String> targetInstanceStateMap) {
+ return targetInstanceStateMap.values().iterator().next().equals(SegmentStateModel.CONSUMING)
Review Comment:
[MAJOR] Do not infer the segment state from the first replica. `getNextSingleSegmentAssignment()` can retain an old replica's `OFFLINE` state while adding target replicas as `CONSUMING`, producing a valid intermediate map such as `{server1=OFFLINE, server3=CONSUMING}`. Because this map is a `TreeMap`, this line can read `OFFLINE` first and miss a consuming segment that is about to move. With `forceCommit=true`, the rebalance then skips the force commit and moves that segment, risking re-consumption or peer-download failure. Please retain the previous `no ONLINE && any CONSUMING` classification (including the null guard), or scan all values, and restore the mixed-state regression case removed in this PR.
--
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] Skip full target recompute on version change for strict realtime tier relocation [pinot]
Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #19054:
URL: https://github.com/apache/pinot/pull/19054#issuecomment-5053506076
## [Codecov](https://app.codecov.io/gh/apache/pinot/pull/19054?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.86%. Comparing base ([`86535a9`](https://app.codecov.io/gh/apache/pinot/commit/86535a9b6b3ec5e959bff9fb2d8cc14f59fc68aa?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)) to head ([`ee0e801`](https://app.codecov.io/gh/apache/pinot/commit/ee0e801a512b838f51c0cb1d0958034bdd216063?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)).
> :exclamation: There is a different number of reports uploaded between BASE (86535a9) and HEAD (ee0e801). Click for more details.
>
> <details><summary>HEAD has 16 uploads less than BASE</summary>
>
>| Flag | BASE (86535a9) | HEAD (ee0e801) |
>|------|------|------|
>|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 #19054 +/- ##
============================================
- Coverage 65.44% 56.86% -8.59%
+ Complexity 1421 1 -1420
============================================
Files 3425 2628 -797
Lines 216146 156483 -59663
Branches 34231 25579 -8652
============================================
- Hits 141452 88979 -52473
+ Misses 63312 59906 -3406
+ Partials 11382 7598 -3784
```
| [Flag](https://app.codecov.io/gh/apache/pinot/pull/19054/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/19054/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/19054/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/19054/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/19054/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/19054/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `56.86% <ø> (-8.59%)` | :arrow_down: |
| [temurin](https://app.codecov.io/gh/apache/pinot/pull/19054/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `56.86% <ø> (-8.59%)` | :arrow_down: |
| [unittests](https://app.codecov.io/gh/apache/pinot/pull/19054/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `56.86% <ø> (-8.58%)` | :arrow_down: |
| [unittests1](https://app.codecov.io/gh/apache/pinot/pull/19054/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `56.86% <ø> (+<0.01%)` | :arrow_up: |
| [unittests2](https://app.codecov.io/gh/apache/pinot/pull/19054/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/19054?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] Skip full target recompute on version change for strict realtime tier relocation [pinot]
Posted by "J-HowHuang (via GitHub)" <gi...@apache.org>.
J-HowHuang commented on code in PR #19054:
URL: https://github.com/apache/pinot/pull/19054#discussion_r3635222411
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/rebalance/TableRebalancer.java:
##########
@@ -585,6 +585,14 @@ private RebalanceResult doRebalance(TableConfig tableConfig, RebalanceConfig reb
new PartitionIdFetcherImpl(tableNameWithType, TableConfigUtils.getPartitionColumn(tableConfig), _helixManager,
isStrictRealtimeSegmentAssignment);
+ // For strict realtime segment assignment, whether this rebalance moves any consuming segment (e.g. rebalancing the
+ // consuming tier) as opposed to only completed segments (e.g. a tier relocation). This is the only case where a
Review Comment:
nit: It has to be tier migration. If it's not migration from `COMPLETED` to other tier, completed segment can't be rebalanced without also rebalancing consuming.
nit:
```suggestion
// consuming tier) as opposed to only completed segments during a tier relocation. This is the only case where a
```
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/rebalance/TableRebalancer.java:
##########
@@ -2233,20 +2248,40 @@ static Set<String> getMovingConsumingSegments(Map<String, Map<String, String>> c
Map<String, Map<String, String>> targetAssignment) {
Set<String> movingConsumingSegments = new HashSet<>();
for (Map.Entry<String, Map<String, String>> entry : currentAssignment.entrySet()) {
- String segmentName = entry.getKey();
- Map<String, String> currentInstanceStateMap = entry.getValue();
- Map<String, String> targetInstanceStateMap = targetAssignment.get(segmentName);
- if (targetInstanceStateMap != null && targetInstanceStateMap.values().stream()
- .noneMatch(state -> state.equals(SegmentStateModel.ONLINE)) && targetInstanceStateMap.values().stream()
- .anyMatch(state -> state.equals(SegmentStateModel.CONSUMING))) {
- if (!currentInstanceStateMap.keySet().equals(targetInstanceStateMap.keySet())) {
- movingConsumingSegments.add(segmentName);
- }
+ if (isMovingConsumingSegment(entry.getValue(), targetAssignment.get(entry.getKey()))) {
+ movingConsumingSegments.add(entry.getKey());
}
}
return movingConsumingSegments;
}
+ /// Returns whether any consuming segment moves between the current and target assignment, short-circuiting on the
+ /// first such segment. Cheaper than `!getMovingConsumingSegments(...).isEmpty()` when only existence is needed
+ /// because it does not materialize the full set.
+ @VisibleForTesting
+ static boolean hasMovingConsumingSegments(Map<String, Map<String, String>> currentAssignment,
+ Map<String, Map<String, String>> targetAssignment) {
+ for (Map.Entry<String, Map<String, String>> entry : currentAssignment.entrySet()) {
+ if (isMovingConsumingSegment(entry.getValue(), targetAssignment.get(entry.getKey()))) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ /// Returns whether the segment is a consuming segment being moved: its target state is `CONSUMING` and its assigned
+ /// instances differ between the current and target assignment. A segment's target instance state map is uniform
+ /// (every replica is `ONLINE`, `CONSUMING`, or `OFFLINE`) because `RealtimeSegmentAssignment` assigns a single state
+ /// to all instances of a moved segment and keeps `OFFLINE` segments unchanged, so the segment's state can be read
+ /// from the first entry. The map is non-null and non-empty: callers iterate the current assignment and look up a
+ /// target derived from it (via `rebalanceTable` / `getNextAssignment`), which covers every current segment, and
+ /// every segment has at least one replica.
+ private static boolean isMovingConsumingSegment(Map<String, String> currentInstanceStateMap,
+ Map<String, String> targetInstanceStateMap) {
+ return targetInstanceStateMap.values().iterator().next().equals(SegmentStateModel.CONSUMING)
Review Comment:
Likely not possible in practice, but should we check any instance is `CONSUMING` instead of the first instance?
--
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] Skip full target recompute on IdealState version change for strict realtime rebalance when instance partitions are unchanged [pinot]
Posted by "Jackie-Jiang (via GitHub)" <gi...@apache.org>.
Jackie-Jiang commented on PR #19054:
URL: https://github.com/apache/pinot/pull/19054#issuecomment-5055473710
Updated the approach once more, refining the previous config-flag gate: the recompute decision now uses the `instancePartitionsUnchanged` result computed by the rebalance itself, instead of the `reassignInstances`/`bootstrap` flags. This way a reassignment that turns out to be a no-op also takes the cheap path. The flag can only ratchet from unchanged to changed within the loop (refreshed by the in-loop instance-partition re-calculation): once the instance partitions changed, the base placements keep moving until the rebalance converges, so the full recompute stays required for the rest of the job. PR description is updated to match.
--
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] Skip full target recompute on IdealState version change for strict realtime rebalance without instance reassignment [pinot]
Posted by "Jackie-Jiang (via GitHub)" <gi...@apache.org>.
Jackie-Jiang commented on code in PR #19054:
URL: https://github.com/apache/pinot/pull/19054#discussion_r3636117612
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/rebalance/TableRebalancer.java:
##########
@@ -2233,20 +2248,40 @@ static Set<String> getMovingConsumingSegments(Map<String, Map<String, String>> c
Map<String, Map<String, String>> targetAssignment) {
Set<String> movingConsumingSegments = new HashSet<>();
for (Map.Entry<String, Map<String, String>> entry : currentAssignment.entrySet()) {
- String segmentName = entry.getKey();
- Map<String, String> currentInstanceStateMap = entry.getValue();
- Map<String, String> targetInstanceStateMap = targetAssignment.get(segmentName);
- if (targetInstanceStateMap != null && targetInstanceStateMap.values().stream()
- .noneMatch(state -> state.equals(SegmentStateModel.ONLINE)) && targetInstanceStateMap.values().stream()
- .anyMatch(state -> state.equals(SegmentStateModel.CONSUMING))) {
- if (!currentInstanceStateMap.keySet().equals(targetInstanceStateMap.keySet())) {
- movingConsumingSegments.add(segmentName);
- }
+ if (isMovingConsumingSegment(entry.getValue(), targetAssignment.get(entry.getKey()))) {
+ movingConsumingSegments.add(entry.getKey());
}
}
return movingConsumingSegments;
}
+ /// Returns whether any consuming segment moves between the current and target assignment, short-circuiting on the
+ /// first such segment. Cheaper than `!getMovingConsumingSegments(...).isEmpty()` when only existence is needed
+ /// because it does not materialize the full set.
+ @VisibleForTesting
+ static boolean hasMovingConsumingSegments(Map<String, Map<String, String>> currentAssignment,
+ Map<String, Map<String, String>> targetAssignment) {
+ for (Map.Entry<String, Map<String, String>> entry : currentAssignment.entrySet()) {
+ if (isMovingConsumingSegment(entry.getValue(), targetAssignment.get(entry.getKey()))) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ /// Returns whether the segment is a consuming segment being moved: its target state is `CONSUMING` and its assigned
+ /// instances differ between the current and target assignment. A segment's target instance state map is uniform
+ /// (every replica is `ONLINE`, `CONSUMING`, or `OFFLINE`) because `RealtimeSegmentAssignment` assigns a single state
+ /// to all instances of a moved segment and keeps `OFFLINE` segments unchanged, so the segment's state can be read
+ /// from the first entry. The map is non-null and non-empty: callers iterate the current assignment and look up a
+ /// target derived from it (via `rebalanceTable` / `getNextAssignment`), which covers every current segment, and
+ /// every segment has at least one replica.
+ private static boolean isMovingConsumingSegment(Map<String, String> currentInstanceStateMap,
+ Map<String, String> targetInstanceStateMap) {
+ return targetInstanceStateMap.values().iterator().next().equals(SegmentStateModel.CONSUMING)
Review Comment:
Good catch, and thanks for the detailed scenario. The PR has been reworked and no longer touches `getMovingConsumingSegments()` at all — the first-entry inference and the removed mixed-state test case are both restored to the original code.
The recompute gate is now based on the rebalance config instead: for strict realtime assignment, the full target recompute on an IdealState version change is skipped only when the rebalance is not reassigning instances (`reassignInstances`/`bootstrap` both false, e.g. the `SegmentRelocator` tier-relocation flow). In that case the instance partitions are read-only for the job, so a segment added mid-rebalance is placed consistently with the target. With instance reassignment, the conservative full recompute is kept, matching prior behavior.
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/rebalance/TableRebalancer.java:
##########
@@ -2233,20 +2248,40 @@ static Set<String> getMovingConsumingSegments(Map<String, Map<String, String>> c
Map<String, Map<String, String>> targetAssignment) {
Set<String> movingConsumingSegments = new HashSet<>();
for (Map.Entry<String, Map<String, String>> entry : currentAssignment.entrySet()) {
- String segmentName = entry.getKey();
- Map<String, String> currentInstanceStateMap = entry.getValue();
- Map<String, String> targetInstanceStateMap = targetAssignment.get(segmentName);
- if (targetInstanceStateMap != null && targetInstanceStateMap.values().stream()
- .noneMatch(state -> state.equals(SegmentStateModel.ONLINE)) && targetInstanceStateMap.values().stream()
- .anyMatch(state -> state.equals(SegmentStateModel.CONSUMING))) {
- if (!currentInstanceStateMap.keySet().equals(targetInstanceStateMap.keySet())) {
- movingConsumingSegments.add(segmentName);
- }
+ if (isMovingConsumingSegment(entry.getValue(), targetAssignment.get(entry.getKey()))) {
+ movingConsumingSegments.add(entry.getKey());
}
}
return movingConsumingSegments;
}
+ /// Returns whether any consuming segment moves between the current and target assignment, short-circuiting on the
+ /// first such segment. Cheaper than `!getMovingConsumingSegments(...).isEmpty()` when only existence is needed
+ /// because it does not materialize the full set.
+ @VisibleForTesting
+ static boolean hasMovingConsumingSegments(Map<String, Map<String, String>> currentAssignment,
+ Map<String, Map<String, String>> targetAssignment) {
+ for (Map.Entry<String, Map<String, String>> entry : currentAssignment.entrySet()) {
+ if (isMovingConsumingSegment(entry.getValue(), targetAssignment.get(entry.getKey()))) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ /// Returns whether the segment is a consuming segment being moved: its target state is `CONSUMING` and its assigned
+ /// instances differ between the current and target assignment. A segment's target instance state map is uniform
+ /// (every replica is `ONLINE`, `CONSUMING`, or `OFFLINE`) because `RealtimeSegmentAssignment` assigns a single state
+ /// to all instances of a moved segment and keeps `OFFLINE` segments unchanged, so the segment's state can be read
+ /// from the first entry. The map is non-null and non-empty: callers iterate the current assignment and look up a
+ /// target derived from it (via `rebalanceTable` / `getNextAssignment`), which covers every current segment, and
+ /// every segment has at least one replica.
+ private static boolean isMovingConsumingSegment(Map<String, String> currentInstanceStateMap,
+ Map<String, String> targetInstanceStateMap) {
+ return targetInstanceStateMap.values().iterator().next().equals(SegmentStateModel.CONSUMING)
Review Comment:
The PR has been reworked and no longer modifies `getMovingConsumingSegments()` — the original `noneMatch(ONLINE)`/`anyMatch(CONSUMING)` classification is untouched. The recompute gate is now based on `reassignInstances`/`bootstrap` instead of consuming-segment movement.
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/rebalance/TableRebalancer.java:
##########
@@ -585,6 +585,14 @@ private RebalanceResult doRebalance(TableConfig tableConfig, RebalanceConfig reb
new PartitionIdFetcherImpl(tableNameWithType, TableConfigUtils.getPartitionColumn(tableConfig), _helixManager,
isStrictRealtimeSegmentAssignment);
+ // For strict realtime segment assignment, whether this rebalance moves any consuming segment (e.g. rebalancing the
+ // consuming tier) as opposed to only completed segments (e.g. a tier relocation). This is the only case where a
Review Comment:
You're right — and your paused-table scenario in the review comment showed the consuming-segment-based gate was insufficient in general (segments uploaded by a minion task mid-rebalance would not be re-collocated). The gate is now based on instance reassignment instead: a newly added segment is placed from the current instance partitions, so as long as the rebalance is not reassigning instances (`reassignInstances`/`bootstrap` both false), it is placed consistently with the target and the cheap path is safe — regardless of whether consuming segments move. With reassignment enabled, the full recompute is kept as before. The comment this nit targeted has been rewritten accordingly.
--
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] Skip full target recompute on IdealState version change for strict realtime rebalance that only moves tier segments [pinot]
Posted by "Jackie-Jiang (via GitHub)" <gi...@apache.org>.
Jackie-Jiang commented on PR #19054:
URL: https://github.com/apache/pinot/pull/19054#issuecomment-5062310823
Reworked the gate based on the discussion above. Instead of keying off `instancePartitionsUnchanged` (which reflects only whether *this* rebalance changed the instance partitions, not whether they were already changed by a prior/interrupted rebalance), the recompute is now skipped only when the rebalance **moves only tier segments** — checked via `providedTierToSegmentsMap`, the authoritative set of tier segments used to build the target assignment.
This is robust to two things raised earlier:
- Tier and non-tier segments are not guaranteed to live on disjoint instances, so I check segment membership rather than comparing target instances against tier instances.
- It doesn't depend on how the instance partitions reached their current state.
When tier segments aren't pre-computed (`updateTargetTier` not run, map null/empty), it conservatively keeps the full recompute. `getMovingConsumingSegments` is left untouched, so the earlier mixed-state concern no longer applies.
--
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] Skip full target recompute on IdealState version change for strict realtime rebalance that only moves tier segments [pinot]
Posted by "Jackie-Jiang (via GitHub)" <gi...@apache.org>.
Jackie-Jiang merged PR #19054:
URL: https://github.com/apache/pinot/pull/19054
--
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