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: [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]
