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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
