This is an automated email from the ASF dual-hosted git repository.
Jackie-Jiang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pinot.git
The following commit(s) were added to refs/heads/master by this push:
new 706b772334e Simplify getMovingConsumingSegments and drop per-segment
stream allocation (#19068)
706b772334e is described below
commit 706b772334e67740187ab7642c73342dbe9e28c2
Author: Xiaotian (Jackie) Jiang <[email protected]>
AuthorDate: Thu Jul 23 15:31:54 2026 -0700
Simplify getMovingConsumingSegments and drop per-segment stream allocation
(#19068)
---
.../helix/core/rebalance/TableRebalancer.java | 25 +++++++++++++++++-----
.../TableRebalancerClusterStatelessTest.java | 10 ++++-----
2 files changed, 25 insertions(+), 10 deletions(-)
diff --git
a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/rebalance/TableRebalancer.java
b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/rebalance/TableRebalancer.java
index 63490ca54a6..648246069fb 100644
---
a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/rebalance/TableRebalancer.java
+++
b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/rebalance/TableRebalancer.java
@@ -2263,6 +2263,12 @@ public class TableRebalancer {
return tierSegments.size() >= segmentsToMove.size() &&
tierSegments.containsAll(segmentsToMove);
}
+ /// Returns the consuming segments that would be moved between the current
and target assignment: those whose target
+ /// has a `CONSUMING` replica and whose assigned instances differ from the
current assignment. Assumes a segment's
+ /// replicas are never a mix of `ONLINE` and `CONSUMING` (a realtime segment
is either consuming or completed, and
+ /// completion flips the whole segment at once), so the scan over a
segment's replicas can stop at the first `ONLINE`
+ /// replica (the segment is completed, hence not consuming) or the first
`CONSUMING` replica (the segment is
+ /// consuming), skipping only `OFFLINE` replicas.
@VisibleForTesting
static Set<String> getMovingConsumingSegments(Map<String, Map<String,
String>> currentAssignment,
Map<String, Map<String, String>> targetAssignment) {
@@ -2271,12 +2277,21 @@ public class TableRebalancer {
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 (targetInstanceStateMap == null) {
+ continue;
+ }
+ for (String state : targetInstanceStateMap.values()) {
+ if (state.equals(SegmentStateModel.ONLINE)) {
+ // Completed segment (no CONSUMING replica by the assumption above),
not a moving consuming segment.
+ break;
+ }
+ if (state.equals(SegmentStateModel.CONSUMING)) {
+ if
(!currentInstanceStateMap.keySet().equals(targetInstanceStateMap.keySet())) {
+ movingConsumingSegments.add(segmentName);
+ }
+ break;
}
+ // OFFLINE replica, keep scanning for an ONLINE or CONSUMING replica.
}
}
return movingConsumingSegments;
diff --git
a/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/rebalance/TableRebalancerClusterStatelessTest.java
b/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/rebalance/TableRebalancerClusterStatelessTest.java
index e42de141c9a..b1f5c6e9deb 100644
---
a/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/rebalance/TableRebalancerClusterStatelessTest.java
+++
b/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/rebalance/TableRebalancerClusterStatelessTest.java
@@ -2424,14 +2424,14 @@ public class TableRebalancerClusterStatelessTest
extends ControllerTest {
tgt3.put("server2", "ONLINE");
targetAssignment.put("segment3", tgt3);
- // Segment 4: one instance is ONLINE, should not be considered
+ // Segment 4: OFFLINE (no CONSUMING replica), should not be considered
Map<String, String> cur4 = new HashMap<>();
- cur4.put("server1", "ONLINE");
- cur4.put("server2", "CONSUMING");
+ cur4.put("server1", "OFFLINE");
+ cur4.put("server2", "OFFLINE");
currentAssignment.put("segment4", cur4);
Map<String, String> tgt4 = new HashMap<>();
- tgt4.put("server1", "ONLINE");
- tgt4.put("server2", "CONSUMING");
+ tgt4.put("server3", "OFFLINE");
+ tgt4.put("server4", "OFFLINE");
targetAssignment.put("segment4", tgt4);
// Segment 5: no ONLINE instance, but at least one in CONSUMING, should be
considered moving
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
