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/10/12 21:57:58 UTC
Re: [PR] Budget-aware low disk mode table rebalance [pinot]
Jackie-Jiang commented on code in PR #19386:
URL: https://github.com/apache/pinot/pull/19386#discussion_r4252685422
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/rebalance/TableRebalancer.java:
##########
@@ -2080,6 +2367,143 @@ private static void updateNumSegmentsToOffloadMap(Map<String, Integer> numSegmen
}
}
+ /// Returns the minimum available replicas the rebalance has to keep up, derived from
+ /// `minReplicasToKeepUpForNoDowntime` and the replication of the segments to be moved. Shared by the rebalance and
+ /// by the pre-checks so that the two cannot drift apart.
+ ///
+ /// NOTE:
+ /// 1. The calculation is based on the number of replicas of the target assignment. In case of increasing the number
+ /// of replicas for the current assignment, the current instance state map might not have enough replicas to reach
+ /// the minimum available replicas requirement. In this scenario we don't want to fail the check, but keep all the
+ /// current instances as this is the best we can do, and can help the table get out of this state.
+ /// 2. Only check the segments to be moved because we don't need to maintain available replicas for segments not
+ /// being moved, including segments with all replicas OFFLINE (error segments during consumption).
+ ///
+ /// @return the minimum available replicas, or [#ILLEGAL_MIN_AVAILABLE_REPLICAS] when
+ /// `minReplicasToKeepUpForNoDowntime` is not less than the replication of the segments to be moved
+ @VisibleForTesting
+ static int getMinAvailableReplicas(Map<String, Map<String, String>> currentAssignment,
+ Map<String, Map<String, String>> targetAssignment, List<String> segmentsToMove,
+ int minReplicasToKeepUpForNoDowntime, Logger tableRebalanceLogger) {
+ int numReplicas = getMinNumReplicas(targetAssignment, segmentsToMove);
+ int minAvailableReplicas;
+ if (minReplicasToKeepUpForNoDowntime >= 0) {
+ // For non-negative value, use it as min available replicas
+ if (minReplicasToKeepUpForNoDowntime >= numReplicas) {
+ return ILLEGAL_MIN_AVAILABLE_REPLICAS;
+ }
+ minAvailableReplicas = minReplicasToKeepUpForNoDowntime;
+ } else {
+ // For negative value, use it as max unavailable replicas
+ minAvailableReplicas = Math.max(numReplicas + minReplicasToKeepUpForNoDowntime, 0);
+ }
+
+ int numCurrentAssignmentReplicas = getMinNumReplicas(currentAssignment, segmentsToMove);
+ if (minAvailableReplicas > numCurrentAssignmentReplicas) {
+ tableRebalanceLogger.warn("minAvailableReplicas: {} larger than existing number of replicas: {}, "
+ + "resetting minAvailableReplicas to {}", minAvailableReplicas, numCurrentAssignmentReplicas,
+ numCurrentAssignmentReplicas);
+ minAvailableReplicas = numCurrentAssignmentReplicas;
+ }
+ return minAvailableReplicas;
+ }
+
+ private static int getMinNumReplicas(Map<String, Map<String, String>> assignment, List<String> segments) {
+ int numReplicas = Integer.MAX_VALUE;
+ for (String segment : segments) {
+ numReplicas = Math.min(assignment.get(segment).size(), numReplicas);
+ }
+ return numReplicas;
+ }
+
+ /// Replays a low disk mode rebalance to find the servers it cannot keep within the disk they start with.
+ ///
+ /// Low disk mode bounds the disk each server may take on to the larger of what it hosts when the rebalance starts
+ /// and what the target assignment places on it. When no progress at all is possible within those bounds - servers
+ /// circularly waiting for one another to drop first - the rebalance relaxes them for a step rather than stalling,
+ /// which is the only way a server can end up over its bound. This runs [#getNextAssignment] over the whole rebalance
+ /// up front to find out whether that happens, and for which servers, before a single segment is moved.
+ ///
+ /// The replay is exact for the assignment it is given, with one caveat: it does not read the segment partition ids,
+ /// so with `batchSizePerServer` enabled and strict replica group routing the segments are grouped more coarsely than
+ /// the rebalance groups them, which changes the pacing of the moves but not the bounds themselves.
+ ///
+ /// @return `null` when the replay could not be finished, which establishes nothing and must not be read as safe;
+ /// an empty map when the rebalance was replayed to the target assignment inside every server's bound; and
+ /// otherwise the servers that would be pushed over their bound, mapped to the bytes they would go over by
+ @Nullable
+ public static Map<String, Long> getServersForcedOverDiskBudget(
+ Map<String, Map<String, String>> currentAssignment, Map<String, Map<String, String>> targetAssignment,
+ int minAvailableReplicas, boolean enableStrictReplicaGroup, int batchSizePerServer,
+ @Nullable TableSizeReader.TableSubTypeSizeDetails tableSizeDetails, Logger tableRebalanceLogger) {
+ return getServersForcedOverDiskBudget(currentAssignment, targetAssignment, minAvailableReplicas,
+ enableStrictReplicaGroup, batchSizePerServer, tableSizeDetails, tableRebalanceLogger,
+ MAX_DISK_BUDGET_REPLAY_STEPS);
+ }
+
+ /// @param maxSteps how many steps to replay before giving up, so that the giving-up path can be tested without
+ /// building an assignment that genuinely needs tens of thousands of steps
+ @Nullable
+ @VisibleForTesting
+ static Map<String, Long> getServersForcedOverDiskBudget(Map<String, Map<String, String>> currentAssignment,
+ Map<String, Map<String, String>> targetAssignment, int minAvailableReplicas, boolean enableStrictReplicaGroup,
+ int batchSizePerServer, @Nullable TableSizeReader.TableSubTypeSizeDetails tableSizeDetails,
+ Logger tableRebalanceLogger, int maxSteps) {
+ DiskUsageBudget diskUsageBudget = DiskUsageBudget.create(currentAssignment, tableSizeDetails);
+ Object2IntOpenHashMap<String> segmentPartitionIdMap = new Object2IntOpenHashMap<>();
+ Map<String, Long> serverToBytesOverBudget = new TreeMap<>();
+ Map<String, Map<String, String>> assignment = currentAssignment;
+ for (int step = 1; step <= maxSteps; step++) {
+ if (assignment.equals(targetAssignment)) {
+ return serverToBytesOverBudget;
+ }
+ Map<String, Long> remainingBytes = diskUsageBudget.forStep(assignment, targetAssignment).getRemainingBytes();
+ Map<String, Map<String, String>> nextAssignment;
+ try {
+ nextAssignment =
+ getNextAssignment(assignment, targetAssignment, minAvailableReplicas, enableStrictReplicaGroup, true,
+ batchSizePerServer, segmentPartitionIdMap, DEFAULT_PARTITION_ID_FETCHER, new NoOpRiskAssessor(),
+ tableRebalanceLogger, diskUsageBudget);
Review Comment:
**[P1] Replay with the same partition IDs used by execution**
`DEFAULT_PARTITION_ID_FETCHER` assigns every segment partition ID 0. With strict routing and batching, collapsing partition boundaries changes allocation decisions and whether the unrestricted fallback is reached. It can therefore return an empty over-budget map while the actual rebalance exceeds the budget.
Using the exact planner methods with `enableStrictReplicaGroup=true`, `minAvailableReplicas=2`, `batchSizePerServer=1`, and all replicas ONLINE, this fixture returns `{}` from the public replay. Running the same assignment with the listed partition IDs overruns h3 by 12 MiB and h6 by 30 MiB. Partitions are cohesive: each partition has a single current/target placement pair.
| Segment | Size (MiB) | Partition | Current hosts | Target hosts |
|---|---:|---:|---|---|
| s007 | 22 | 2 | h5,h7 | h0,h4,h6 |
| s008 | 28 | 3 | h5,h7 | h0,h4,h6 |
| s013 | 37 | 4 | h0,h2,h3 | h2,h4,h6 |
| s016 | 19 | 8 | h0,h4,h6 | h2,h5,h7 |
| s017 | 15 | 10 | h0,h4,h6 | h2,h5,h7 |
| s018 | 24 | 10 | h0,h4,h6 | h2,h5,h7 |
| s019 | 49 | 9 | h0,h4,h6 | h2,h5,h7 |
| s020 | 30 | 9 | h0,h4,h6 | h2,h5,h7 |
| s022 | 20 | 13 | h0,h1,h2 | h0,h3,h6 |
| s023 | 25 | 15 | h0,h1,h2 | h0,h3,h6 |
| s024 | 12 | 13 | h0,h1,h2 | h0,h3,h6 |
h6 starts at 137 MiB, targets 144 MiB, and reaches 174 MiB during execution. For a concrete disk-threshold example, give h6 970 MiB used out of 2,000 MiB at a 50% threshold, and give the other servers ample capacity. With the table's reported per-replica size set to 281 MiB, the pre-check estimates h6 below threshold after rebalance and invokes replay for the transient risk. Replay's empty result produces PASS, but execution reaches 1,007 MiB used, above the 1,000 MiB threshold.
Please supply the real partition IDs to replay, or report this configuration as inconclusive. A regression should compare replay with partition-aware execution for this fixture.
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/rebalance/TableRebalancer.java:
##########
@@ -1687,19 +1695,266 @@ static Map<String, Map<String, String>> getNextAssignment(Map<String, Map<String
private static Map<String, Map<String, String>> getNextAssignment(Map<String, Map<String, String>> currentAssignment,
Map<String, Map<String, String>> targetAssignment, int minAvailableReplicas, boolean enableStrictReplicaGroup,
boolean lowDiskMode, int batchSizePerServer, Object2IntOpenHashMap<String> segmentPartitionIdMap,
- PartitionIdFetcher partitionIdFetcher, DataLossRiskAssessor dataLossRiskAssessor, Logger tableRebalanceLogger) {
+ PartitionIdFetcher partitionIdFetcher, DataLossRiskAssessor dataLossRiskAssessor, Logger tableRebalanceLogger,
+ @Nullable DiskUsageBudget diskUsageBudget) {
+ if (!lowDiskMode) {
+ return computeNextAssignment(currentAssignment, targetAssignment, minAvailableReplicas, enableStrictReplicaGroup,
+ false, batchSizePerServer, segmentPartitionIdMap, partitionIdFetcher, dataLossRiskAssessor,
+ tableRebalanceLogger, null);
+ }
+
+ // We use DiskUsageBudget and StepDiskBudget here to keep track of per-server byte gain/loss and to guarantee the
+ // invariant of lowDiskMode where each server won't get more bytes than its net gain at any intermediate steps.
+ Preconditions.checkState(diskUsageBudget != null, "Low disk mode requires a disk usage budget");
+ StepDiskBudget stepBudget = diskUsageBudget.forStep(currentAssignment, targetAssignment);
+ Map<String, Map<String, String>> nextAssignment =
+ computeNextAssignment(currentAssignment, targetAssignment, minAvailableReplicas, enableStrictReplicaGroup, true,
+ batchSizePerServer, segmentPartitionIdMap, partitionIdFetcher, dataLossRiskAssessor, tableRebalanceLogger,
+ stepBudget);
+ if (!nextAssignment.equals(currentAssignment)) {
+ return nextAssignment;
+ }
+
+ // Can't progress while respecting the budget. This could happen if servers circular wait for others to drop first.
+ tableRebalanceLogger.warn("Cannot make progress in low disk mode without exceeding the disk usage a server "
+ + "started the rebalance with. Allowing the additions for this step, which temporarily increases the disk "
+ + "usage of these servers. Bytes each server could still take on: {}", stepBudget.getRemainingBytes());
+ return computeNextAssignment(currentAssignment, targetAssignment, minAvailableReplicas, enableStrictReplicaGroup,
+ true, batchSizePerServer, segmentPartitionIdMap, partitionIdFetcher, dataLossRiskAssessor,
+ tableRebalanceLogger, null);
+ }
+
+ /// @param stepBudget bounds the disk each server may take on in this step, `null` to apply no bound
+ private static Map<String, Map<String, String>> computeNextAssignment(
+ Map<String, Map<String, String>> currentAssignment, Map<String, Map<String, String>> targetAssignment,
+ int minAvailableReplicas, boolean enableStrictReplicaGroup, boolean lowDiskMode, int batchSizePerServer,
+ Object2IntOpenHashMap<String> segmentPartitionIdMap, PartitionIdFetcher partitionIdFetcher,
+ DataLossRiskAssessor dataLossRiskAssessor, Logger tableRebalanceLogger,
+ @Nullable StepDiskBudget stepBudget) {
return enableStrictReplicaGroup
? getNextStrictReplicaGroupAssignment(currentAssignment, targetAssignment, minAvailableReplicas, lowDiskMode,
- batchSizePerServer, segmentPartitionIdMap, partitionIdFetcher, dataLossRiskAssessor, tableRebalanceLogger)
+ batchSizePerServer, segmentPartitionIdMap, partitionIdFetcher, dataLossRiskAssessor, tableRebalanceLogger,
+ stepBudget)
: getNextNonStrictReplicaGroupAssignment(currentAssignment, targetAssignment, minAvailableReplicas,
- lowDiskMode, batchSizePerServer, dataLossRiskAssessor);
+ lowDiskMode, batchSizePerServer, dataLossRiskAssessor, stepBudget);
+ }
+
+ /// Bounds the disk a server may use while a low disk mode rebalance is in flight.
+ ///
+ /// If a server results in net byte loss, its ceiling is as many bytes as it originally hosted.
+ /// If a server results in net byte gain, its ceiling is as many bytes as it will eventually host.
+ ///
+ /// This invariant helps us guarantee that a rebalance won't introduce bytes more than necessary on the disk at any
+ /// moment during the rebalance. Thus, rebalance can safely run as long as we know the target assignment won't go
+ /// beyond their disk utilization (this can be guard by either pre-check or resource utilization checker etc.)
+ ///
+ /// In the case when new segments appear in ideal state by external sources (e.g. segment upload, consuming segment
+ /// committed), there are two cases:
+ /// 1. They are added to the ideal state as the target assignment. This case the rebalance steps outcome won't change
+ /// 2. They are added to the ideal state that's different to their target assignments. For example when strict
+ /// replica routing is in place. This case, it might lead to a different computeNextAssignment result.
+ ///
+ /// When no per-segment size is known every segment counts as one byte, which degenerates to bounding the number of
+ /// segments hosted rather than the bytes.
+ @VisibleForTesting
+ static class DiskUsageBudget {
+ private final Map<String, Long> _segmentSizeBytes;
+ private final long _defaultSegmentSizeBytes;
+ private final Map<String, Long> _anchorServerBytes;
+ private final Set<String> _accountedSegments;
+
+ private DiskUsageBudget(Map<String, Long> segmentSizeBytes, long defaultSegmentSizeBytes,
+ Map<String, Long> anchorServerBytes, Set<String> accountedSegments) {
+ _segmentSizeBytes = segmentSizeBytes;
+ _defaultSegmentSizeBytes = defaultSegmentSizeBytes;
+ _anchorServerBytes = anchorServerBytes;
+ _accountedSegments = accountedSegments;
+ }
+
+ /// @param initialAssignment the assignment the rebalance starts from, used to anchor the per-server ceiling
+ /// @param tableSizeDetails per-segment sizes, `null` to fall back to counting segments instead of bytes
+ static DiskUsageBudget create(Map<String, Map<String, String>> initialAssignment,
+ @Nullable TableSizeReader.TableSubTypeSizeDetails tableSizeDetails) {
+ Map<String, Long> segmentSizeBytes = new HashMap<>();
+ long totalKnownBytes = 0;
+ if (tableSizeDetails != null) {
+ for (Map.Entry<String, TableSizeReader.SegmentSizeDetails> entry : tableSizeDetails._segments.entrySet()) {
+ // The size a single replica takes up on disk, which is what one server pays for hosting the segment
+ long sizeBytes = entry.getValue()._maxReportedSizePerReplicaInBytes;
+ if (sizeBytes > 0) {
+ segmentSizeBytes.put(entry.getKey(), sizeBytes);
+ totalKnownBytes += sizeBytes;
+ }
+ }
+ }
+ // Segments whose size could not be read (missing from all servers, or still consuming) are charged the average
+ // size of the segments that could be read. When no per-segment size is available at all, fall back to the
+ // average over the whole table, which is the same estimate the disk utilization pre-check works with. Failing
+ // that, charge every segment one byte, which turns the budget into a bound on the number of segments hosted
+ long defaultSegmentSizeBytes;
+ if (!segmentSizeBytes.isEmpty()) {
+ defaultSegmentSizeBytes = Math.max(1L, totalKnownBytes / segmentSizeBytes.size());
+ } else if (tableSizeDetails != null && tableSizeDetails._reportedSizePerReplicaInBytes > 0
+ && !initialAssignment.isEmpty()) {
+ defaultSegmentSizeBytes =
+ Math.max(1L, tableSizeDetails._reportedSizePerReplicaInBytes / initialAssignment.size());
+ } else {
+ defaultSegmentSizeBytes = 1L;
+ }
+ Map<String, Long> initialServerBytes = new HashMap<>();
+ for (Map.Entry<String, Map<String, String>> entry : initialAssignment.entrySet()) {
+ long sizeBytes = segmentSizeBytes.getOrDefault(entry.getKey(), defaultSegmentSizeBytes);
+ for (String instance : entry.getValue().keySet()) {
+ initialServerBytes.merge(instance, sizeBytes, Long::sum);
+ }
+ }
+ return new DiskUsageBudget(segmentSizeBytes, defaultSegmentSizeBytes, initialServerBytes,
+ new HashSet<>(initialAssignment.keySet()));
+ }
+
+ long getSegmentSizeBytes(String segmentName) {
+ return _segmentSizeBytes.getOrDefault(segmentName, _defaultSegmentSizeBytes);
+ }
+
+ /// Adds the segments that appeared since the anchor was taken - an uploaded segment, or a new consuming segment -
+ /// to the anchor of the servers hosting them.
+ ///
+ /// With this, this segment wouldn't be account for the budget if it doesn't need to be moved at all.
+ private void accountForNewSegments(Map<String, Map<String, String>> currentAssignment) {
+ for (Map.Entry<String, Map<String, String>> entry : currentAssignment.entrySet()) {
+ if (_accountedSegments.add(entry.getKey())) {
+ long sizeBytes = getSegmentSizeBytes(entry.getKey());
+ for (String instance : entry.getValue().keySet()) {
+ _anchorServerBytes.merge(instance, sizeBytes, Long::sum);
+ }
+ }
+ }
+ }
+
+ Map<String, Long> getServerToHostedBytes(Map<String, Map<String, String>> assignment) {
+ Map<String, Long> serverToHostedBytes = new HashMap<>();
+ for (Map.Entry<String, Map<String, String>> entry : assignment.entrySet()) {
+ long sizeBytes = getSegmentSizeBytes(entry.getKey());
+ for (String instance : entry.getValue().keySet()) {
+ serverToHostedBytes.merge(instance, sizeBytes, Long::sum);
+ }
+ }
+ return serverToHostedBytes;
+ }
+
+ /// Totals the budgeted bytes of the segments sharing each pair of current and target instances, over whichever
+ /// part of the assignment it is given. Those segments are all assigned the same instances, so an instance added
+ /// for one of them takes on all of them.
+ Map<Pair<Set<String>, Set<String>>, Long> getInstancePairToBytes(
+ Map<String, Map<String, String>> currentAssignment, Map<String, Map<String, String>> targetAssignment) {
+ Map<Pair<Set<String>, Set<String>>, Long> instancePairToBytes = new HashMap<>();
+ for (Map.Entry<String, Map<String, String>> entry : currentAssignment.entrySet()) {
+ Map<String, String> targetInstanceStateMap = targetAssignment.get(entry.getKey());
+ if (targetInstanceStateMap != null) {
+ instancePairToBytes.merge(Pair.of(entry.getValue().keySet(), targetInstanceStateMap.keySet()),
+ getSegmentSizeBytes(entry.getKey()), Long::sum);
+ }
+ }
+ return instancePairToBytes;
+ }
+
+ /// Returns the bytes each server may still take on in this step, i.e. `ceiling - currently hosted`.
+ StepDiskBudget forStep(Map<String, Map<String, String>> currentAssignment,
+ Map<String, Map<String, String>> targetAssignment) {
+ accountForNewSegments(currentAssignment);
+ Map<String, Long> hostedBytes = getServerToHostedBytes(currentAssignment);
+ Map<String, Long> targetBytes = getServerToHostedBytes(targetAssignment);
+ Set<String> servers = new HashSet<>(_anchorServerBytes.keySet());
+ servers.addAll(hostedBytes.keySet());
+ servers.addAll(targetBytes.keySet());
+ Map<String, Long> remainingBytes = new HashMap<>();
+ for (String server : servers) {
+ long ceilingBytes =
+ Math.max(_anchorServerBytes.getOrDefault(server, 0L), targetBytes.getOrDefault(server, 0L));
+ remainingBytes.put(server, Math.max(0L, ceilingBytes - hostedBytes.getOrDefault(server, 0L)));
+ }
+ // Segments that share a current and target instance pair are all assigned the same instances, so a newly added
+ // instance takes on all of them. Total their size up so that the instances to add can be picked with that in
+ // mind rather than only on the number of segments to offload
+ return new StepDiskBudget(this, remainingBytes,
+ getInstancePairToBytes(currentAssignment, targetAssignment));
+ }
+ }
+
+ /// The [DiskUsageBudget] left for a single rebalance step, drawn down as segments are assigned.
+ @VisibleForTesting
+ static class StepDiskBudget {
+ private final DiskUsageBudget _budget;
+ private final Map<String, Long> _remainingBytes;
+ // this is to record for each strict replica group, how many bytes they contribute altogether since they'll be
+ // moved together
+ private final Map<Pair<Set<String>, Set<String>>, Long> _instancePairToRequiredBytes;
+
+ private StepDiskBudget(DiskUsageBudget budget, Map<String, Long> remainingBytes,
+ Map<Pair<Set<String>, Set<String>>, Long> instancePairToRequiredBytes) {
+ _budget = budget;
+ _remainingBytes = remainingBytes;
+ _instancePairToRequiredBytes = instancePairToRequiredBytes;
+ }
+
+ /// Moves the instances that cannot take on all the segments sharing this current and target instance pair to the
+ /// back of the list so that they will be picked as the next step first.
+ ///
+ /// The instances that do not fit are kept at the back rather than dropped, so that an assignment is still produced
+ /// when none of them fits and the budget stays the only thing that decides whether it is applied.
+ List<Triple<String, String, Integer>> retainInstancesThatFit(
+ List<Triple<String, String, Integer>> instancesInfo, Set<String> currentInstances,
+ Set<String> targetInstances) {
+ long requiredBytes = _instancePairToRequiredBytes.getOrDefault(Pair.of(currentInstances, targetInstances), 0L);
+ List<Triple<String, String, Integer>> fits = new ArrayList<>(instancesInfo.size());
+ for (Triple<String, String, Integer> instanceInfo : instancesInfo) {
+ if (_remainingBytes.getOrDefault(instanceInfo.getLeft(), 0L) >= requiredBytes) {
+ fits.add(instanceInfo);
Review Comment:
**[P1] Use segment-sized admission checks in the non-strict path**
This filter is shared with `getNextNonStrictReplicaGroupAssignment()` through `getNextSingleSegmentAssignment()`. It requires room for every segment sharing the current/target instance pair, although the non-strict caller applies batching and disk charging independently to each segment. The shared `assignmentMap` keeps candidate destinations consistent; it does not make adoption atomic. Whole-pair sizing is appropriate for strict, unbatched movement, but it can reject legal individual moves here.
This causes an avoidable unsafe fallback. With `enableStrictReplicaGroup=false`, `minAvailableReplicas=3`, `batchSizePerServer=1`, and all replicas ONLINE, the following five-segment assignment reproduces it. Host numbers denote distinct servers; sizes can be interpreted as MiB.
| Segment | Size | Current hosts | Target hosts |
|---|---:|---|---|
| segment007 | 3 | 0,1,2,3 | 2,4,5,6 |
| segment017 | 1 | 2,4,5,6 | 0,1,3,4 |
| segment018 | 4 | 2,4,5,6 | 0,1,3,4 |
| segment031 | 1 | 1,2,3,4 | 0,2,5,6 |
| segment032 | 1 | 1,2,3,4 | 0,2,5,6 |
At step 5, segment017 needs 1 unit, and hosts 0, 1, and 3 have 2, 4, and 4 units available. This filter nevertheless requires 5 units for segment017 plus segment018, rejecting all three destinations. The bounded pass stalls and the unrestricted fallback adds 3 units to host5 and 1 to host6, both with zero budget remaining. Changing only the non-strict prefilter's required amount to the current segment's size in an isolated comparison reaches the target without any overrun.
Please size the admission check to the unit the caller actually accepts or defers: a segment in this path, and the relevant partition group for strict batching. Preserve whole-group admission where movement is atomic.
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/rebalance/DefaultRebalancePreChecker.java:
##########
@@ -329,10 +333,70 @@ protected RebalancePreCheckerResult checkDiskUtilization(PreCheckContext preChec
}
String serversGoingOver = " Servers that would go over it DURING the rebalance: " + String.join(", ",
serversUnsafeDuringRebalance) + ".";
+ // lowDiskMode bounds the disk each server takes on to what it starts the rebalance with, but when the rebalance
+ // cannot progress at all within those bounds it relaxes them for a step rather than stalling. Replay the rebalance
+ // to find out whether it has to, instead of assuming lowDiskMode always avoids the transient usage
+ Map<String, Long> serversForcedOverBudget = getServersForcedOverDiskBudget(preCheckContext);
+ String unsafeDuringRebalance =
+ getUnsafeDiskUtilizationMessage("DURING rebalance", serversUnsafeDuringRebalance, threshold);
+ if (serversForcedOverBudget == null) {
+ // The replay could not be finished due to exception or step limit, so nothing was established. This is expected
+ // to be rare.
+ return RebalancePreCheckerResult.error(unsafeDuringRebalance + ". Whether lowDiskMode can avoid it for this "
+ + "target assignment could not be established, see the controller log. Treat the transient disk usage above "
+ + "as unsafe");
+ }
+ if (!serversForcedOverBudget.isEmpty()) {
+ return RebalancePreCheckerResult.error(unsafeDuringRebalance
+ + ". lowDiskMode cannot avoid it for this target assignment: the rebalance cannot make progress without "
+ + "going over the disk these servers start with, by up to " + formatBytesOverBudget(serversForcedOverBudget)
+ + ". Rebalance to a target assignment that frees up space on them first, or add capacity");
Review Comment:
**[P2] Compare the replayed peak with the disk-utilization threshold**
A nonempty replay result establishes that the per-table start/target byte budget is exceeded; it does not establish that a server crosses the configured disk-utilization threshold. This branch nevertheless returns ERROR using the earlier worst-case utilization estimate.
The new `testOverThresholdDuringRebalanceIsAnErrorWhenLowDiskModeCannotAvoidIt` demonstrates the false alarm: host05 starts with 7,800 MiB used out of 20,000 MiB, and the exact replay adds at most 31 MiB above its initial hosted bytes. Its actual peak is therefore 7,831 MiB (39.155%), safely below the 50% / 10,000 MiB threshold, although the test expects ERROR and recommends adding capacity.
Please compare each server's replayed peak, combined with its existing disk usage, against its threshold headroom. Keep a budget-constraint warning separate if useful, and add a case where budget excess remains within the disk threshold.
--
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