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