diff --git a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/rebalance/DefaultRebalancePreChecker.java b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/rebalance/DefaultRebalancePreChecker.java index 50bb56849efb..e99906b6ebb6 100644 --- a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/rebalance/DefaultRebalancePreChecker.java +++ b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/rebalance/DefaultRebalancePreChecker.java @@ -56,8 +56,7 @@ public class DefaultRebalancePreChecker implements RebalancePreChecker { public static final String NEEDS_RELOAD_STATUS = "needsReloadStatus"; public static final String IS_MINIMIZE_DATA_MOVEMENT = "isMinimizeDataMovement"; - public static final String DISK_UTILIZATION_DURING_REBALANCE = "diskUtilizationDuringRebalance"; - public static final String DISK_UTILIZATION_AFTER_REBALANCE = "diskUtilizationAfterRebalance"; + public static final String DISK_UTILIZATION = "diskUtilization"; public static final String REBALANCE_CONFIG_OPTIONS = "rebalanceConfigOptions"; public static final String REPLICA_GROUPS_INFO = "replicaGroupsInfo"; @@ -106,13 +105,10 @@ public Map check(PreCheckContext preCheckCont diskUtilizationThreshold = 1.0; } - // Check if all servers involved in the rebalance have enough disk space for rebalance operation. + // Check if all servers involved in the rebalance have enough disk space, both while the rebalance is running and + // once it is done. // Notice this check could have false positives (disk utilization is subject to change by other operations anytime) - preCheckResult.put(DISK_UTILIZATION_DURING_REBALANCE, - checkDiskUtilization(preCheckContext, diskUtilizationThreshold, true)); - // Check if all servers involved in the rebalance will have enough disk space after the rebalance. - preCheckResult.put(DISK_UTILIZATION_AFTER_REBALANCE, - checkDiskUtilization(preCheckContext, diskUtilizationThreshold, false)); + preCheckResult.put(DISK_UTILIZATION, checkDiskUtilization(preCheckContext, diskUtilizationThreshold)); preCheckResult.put(REBALANCE_CONFIG_OPTIONS, checkRebalanceConfig(rebalanceConfig, tableConfig, preCheckContext.getCurrentAssignment(), preCheckContext.getTargetAssignment(), @@ -269,19 +265,31 @@ private RebalancePreCheckerResult checkIsMinimizeDataMovement(TableConfig tableC } /// Estimates whether the servers of the target assignment stay within the disk utilization threshold, based on the - /// average segment size and the number of segments added to (and, unless checking for the worst case, removed from) - /// each server. Every segment is assumed to take up disk space on each server it is assigned to. Downstream projects - /// where that does not hold (e.g. because a segment can be stored outside of the server, as indicated by + /// average segment size and the number of segments added to and removed from each server. Two points in time are + /// estimated: + /// + /// - **After the rebalance**, i.e. once every server has both added and removed all the segments it has to. Going + /// over the threshold there is an error whatever the rebalance config, since no way of running the rebalance + /// brings the end state back within the threshold. + /// - **During the rebalance**, where a server can transiently hold the segments it is gaining on top of the ones it + /// is about to lose. Only servers actually gaining segments are estimated: one that merely sheds them can only be + /// over the threshold because it already was, which the rebalance neither causes nor can be blamed for. Only + /// `lowDiskMode` rules that peak out, by waiting for the segments to be deleted before adding the new ones, so + /// going over the threshold there is an error unless it is enabled. `downtime` does not help: it replaces the + /// IdealState with the target assignment in one go, without ordering the drops before the adds. Worse, that + /// one-shot path skips the incremental one `lowDiskMode` acts on, so `downtime` cancels `lowDiskMode` out + /// entirely. `bestEfforts` weakens it rather than cancelling it — the deletes are still awaited, just no longer + /// unconditionally — so it downgrades the result to a warning instead of an error. + /// + /// Every segment is assumed to take up disk space on each server it is assigned to. Downstream projects where that + /// does not hold (e.g. because a segment can be stored outside of the server, as indicated by /// [TierConfig#getTierBackend()]) can override this. - protected RebalancePreCheckerResult checkDiskUtilization(PreCheckContext preCheckContext, double threshold, - boolean worstCase) { + protected RebalancePreCheckerResult checkDiskUtilization(PreCheckContext preCheckContext, double threshold) { Map> currentAssignment = preCheckContext.getCurrentAssignment(); Map> targetAssignment = preCheckContext.getTargetAssignment(); TableSizeReader.TableSubTypeSizeDetails tableSubTypeSizeDetails = preCheckContext.getTableSubTypeSizeDetails(); - boolean isDiskUtilSafe = true; - StringBuilder message = - new StringBuilder("UNSAFE. Servers with unsafe disk utilization (>" + (short) (threshold * 100) + "%): "); - String sep = ""; + List serversUnsafeDuringRebalance = new ArrayList<>(); + List serversUnsafeAfterRebalance = new ArrayList<>(); Map> existingServersToSegmentMap = new HashMap<>(); Map> newServersToSegmentMap = new HashMap<>(); @@ -327,22 +335,60 @@ protected RebalancePreCheckerResult checkDiskUtilization(PreCheckContext preChec long diskUtilizationGain = newSegmentSet.size() * avgSegmentSize; long diskUtilizationLoss = removedSegmentSet.size() * avgSegmentSize; - long diskUtilizationFootprint = - diskUsage.getUsedSpaceBytes() + diskUtilizationGain - (worstCase ? 0 : diskUtilizationLoss); - double diskUtilizationFootprintRatio = - (double) diskUtilizationFootprint / diskUsage.getTotalSpaceBytes(); - - if (diskUtilizationFootprintRatio >= threshold) { - isDiskUtilSafe = false; - message.append(sep) - .append(server) - .append(String.format(" (%d%%)", (short) (diskUtilizationFootprintRatio * 100))); - sep = ", "; + // While the rebalance is running, the segments being added can co-exist with the ones being removed. A server + // gaining nothing never builds up that transient usage: it is only ever over the threshold because it already + // was, which is not something this rebalance causes nor something lowDiskMode could do anything about. If it is + // still over once the rebalance is done, the estimate below catches it + if (diskUtilizationGain > 0) { + addIfOverThreshold(serversUnsafeDuringRebalance, server, + (double) (diskUsage.getUsedSpaceBytes() + diskUtilizationGain) / diskUsage.getTotalSpaceBytes(), threshold); } + addIfOverThreshold(serversUnsafeAfterRebalance, server, + (double) (diskUsage.getUsedSpaceBytes() + diskUtilizationGain - diskUtilizationLoss) + / diskUsage.getTotalSpaceBytes(), threshold); } - return isDiskUtilSafe ? RebalancePreCheckerResult.pass( - String.format("Within threshold (<%d%%)", (short) (threshold * 100))) - : RebalancePreCheckerResult.error(message.toString()); + + // A server over the threshold once the rebalance is done is over it during the rebalance as well, so the end state + // is what to report first: it is both the more severe problem and the one that has to be solved by adding capacity + // rather than by tuning the rebalance config + if (!serversUnsafeAfterRebalance.isEmpty()) { + return RebalancePreCheckerResult.error( + getUnsafeDiskUtilizationMessage("AFTER rebalance", serversUnsafeAfterRebalance, threshold)); + } + String withinThreshold = String.format("Within threshold (<%d%%)", (short) (threshold * 100)); + if (serversUnsafeDuringRebalance.isEmpty()) { + return RebalancePreCheckerResult.pass(withinThreshold); + } + // lowDiskMode is the only way to rule the transient disk usage above out, since it waits for the segments to be + // deleted before adding the new ones. It is however only honored by the incremental rebalance path, which downtime + // skips altogether by replacing the IdealState with the target assignment in one go + RebalanceConfig rebalanceConfig = preCheckContext.getRebalanceConfig(); + if (rebalanceConfig.isDowntime() || !rebalanceConfig.isLowDiskMode()) { + return RebalancePreCheckerResult.error( + getUnsafeDiskUtilizationMessage("DURING rebalance", serversUnsafeDuringRebalance, threshold) + + (rebalanceConfig.isDowntime() + ? ". lowDiskMode, which would delete segments before adding the new ones, has no effect while downtime " + + "is enabled" + : ". Enable lowDiskMode to delete segments before adding the new ones")); + } + String serversGoingOver = " Servers that would go over it DURING the rebalance: " + String.join(", ", + serversUnsafeDuringRebalance) + "."; + return RebalancePreCheckerResult.pass(withinThreshold + " AFTER rebalance." + serversGoingOver + " lowDiskMode " + + "avoids that transient disk usage by deleting segments before adding the new ones"); + } + + private static void addIfOverThreshold(List servers, String server, double utilizationRatio, + double threshold) { + if (utilizationRatio >= threshold) { + servers.add(server + String.format(" (%d%%)", (short) (utilizationRatio * 100))); + } + } + + /// The threshold is rendered as `>=` because [#addIfOverThreshold] flags a server whose utilization reaches it, not + /// only one that exceeds it. + private static String getUnsafeDiskUtilizationMessage(String when, List servers, double threshold) { + return String.format("UNSAFE. Servers with unsafe disk utilization %s (>=%d%%): %s", when, + (short) (threshold * 100), String.join(", ", servers)); } private RebalancePreCheckerResult checkRebalanceConfig(RebalanceConfig rebalanceConfig, TableConfig tableConfig, @@ -372,6 +418,13 @@ private RebalancePreCheckerResult checkRebalanceConfig(RebalanceConfig rebalance pass = false; warnings.add("Number of replicas (" + numReplicas + ") is greater than 1, downtime is not recommended."); } + // Downtime replaces the IdealState with the target assignment in one go, skipping the incremental path that is + // the only one honoring lowDiskMode + if (rebalanceConfig.isLowDiskMode()) { + pass = false; + warnings.add("lowDiskMode has no effect when downtime is enabled, disable downtime for segments to be deleted " + + "before the new ones are added."); + } } // Peer download enabled tables may have data loss during rebalance, when downtime=true or minAvailableReplicas=0. diff --git a/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/rebalance/DefaultRebalancePreCheckerTest.java b/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/rebalance/DefaultRebalancePreCheckerTest.java new file mode 100644 index 000000000000..ecf5583965a2 --- /dev/null +++ b/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/rebalance/DefaultRebalancePreCheckerTest.java @@ -0,0 +1,216 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.pinot.controller.helix.core.rebalance; + +import java.util.Map; +import org.apache.pinot.common.restlet.resources.DiskUsageInfo; +import org.apache.pinot.common.restlet.resources.RebalanceConfig; +import org.apache.pinot.common.restlet.resources.RebalancePreCheckerResult; +import org.apache.pinot.common.restlet.resources.RebalancePreCheckerResult.PreCheckStatus; +import org.apache.pinot.controller.helix.core.rebalance.RebalancePreChecker.PreCheckContext; +import org.apache.pinot.controller.util.TableSizeReader; +import org.apache.pinot.controller.validation.ResourceUtilizationInfo; +import org.apache.pinot.spi.config.table.TableConfig; +import org.apache.pinot.spi.config.table.TableType; +import org.apache.pinot.spi.utils.builder.TableConfigBuilder; +import org.testng.annotations.AfterClass; +import org.testng.annotations.Test; + +import static org.apache.pinot.spi.utils.CommonConstants.Helix.StateModel.SegmentStateModel.ONLINE; +import static org.testng.Assert.assertEquals; +import static org.testng.Assert.assertTrue; + + +/// Tests the consolidated disk utilization pre-check of [DefaultRebalancePreChecker]. +/// +/// Unless stated otherwise, the tests share the same assignment: 4 segments of 100 bytes each, moving so that both +/// servers gain and shed some. `Server_0` gains 100 bytes and sheds 200, `Server_1` gains 200 and sheds 100. Every +/// server has 1000 bytes of total space and the threshold is 50% of it. +public class DefaultRebalancePreCheckerTest { + private static final String SERVER_0 = "Server_0"; + private static final String SERVER_1 = "Server_1"; + private static final long TOTAL_SPACE_BYTES = 1000L; + private static final long TABLE_SIZE_PER_REPLICA_BYTES = 400L; + private static final double THRESHOLD = 0.5; + + private static final Map> CURRENT_ASSIGNMENT = + Map.of("segment_0", Map.of(SERVER_0, ONLINE), "segment_1", Map.of(SERVER_0, ONLINE), "segment_2", + Map.of(SERVER_0, ONLINE), "segment_3", Map.of(SERVER_1, ONLINE)); + private static final Map> TARGET_ASSIGNMENT = + Map.of("segment_0", Map.of(SERVER_0, ONLINE), "segment_1", Map.of(SERVER_1, ONLINE), "segment_2", + Map.of(SERVER_1, ONLINE), "segment_3", Map.of(SERVER_0, ONLINE)); + + /// An assignment where `Server_0` only sheds segments: it gives its 2 remaining segments to `Server_1` and gains + /// nothing. + private static final Map> SHED_ONLY_CURRENT_ASSIGNMENT = + Map.of("segment_0", Map.of(SERVER_0, ONLINE), "segment_1", Map.of(SERVER_0, ONLINE), "segment_2", + Map.of(SERVER_0, ONLINE), "segment_3", Map.of(SERVER_0, ONLINE)); + private static final Map> SHED_ONLY_TARGET_ASSIGNMENT = + Map.of("segment_0", Map.of(SERVER_0, ONLINE), "segment_1", Map.of(SERVER_0, ONLINE), "segment_2", + Map.of(SERVER_1, ONLINE), "segment_3", Map.of(SERVER_1, ONLINE)); + + private final DefaultRebalancePreChecker _preChecker = new DefaultRebalancePreChecker(); + + /// [ResourceUtilizationInfo] is a mutable static shared by everything running in the same JVM fork. + @AfterClass + public void tearDown() { + ResourceUtilizationInfo.setDiskUsageInfo(Map.of()); + } + + @Test + public void testWithinThresholdBothDuringAndAfterRebalance() { + // Server_0 peaks at 400 of its 1000 bytes and Server_1 at 300, neither reaches 500 + setDiskUsage(300L, 100L); + RebalancePreCheckerResult result = checkDiskUtilization(new RebalanceConfig()); + assertEquals(result.getPreCheckStatus(), PreCheckStatus.PASS); + assertEquals(result.getMessage(), "Within threshold (<50%)"); + } + + @Test + public void testOverThresholdAfterRebalanceIsAnErrorWhateverTheRebalanceConfig() { + // Server_1 ends up at 520 of its 1000 bytes, which no rebalance config can bring back under the threshold + setDiskUsage(100L, 420L); + for (RebalanceConfig rebalanceConfig : new RebalanceConfig[]{ + new RebalanceConfig(), lowDiskMode(), downtime(), bestEfforts() + }) { + RebalancePreCheckerResult result = checkDiskUtilization(rebalanceConfig); + assertEquals(result.getPreCheckStatus(), PreCheckStatus.ERROR); + assertEquals(result.getMessage(), + "UNSAFE. Servers with unsafe disk utilization AFTER rebalance (>=50%): " + SERVER_1 + " (52%)"); + } + } + + @Test + public void testOverThresholdOnlyDuringRebalanceIsAnErrorWithoutLowDiskMode() { + // Server_1 transiently holds the 200 bytes it gains on top of the 100 it is about to shed, peaking at 520 + setDiskUsage(100L, 320L); + RebalancePreCheckerResult result = checkDiskUtilization(new RebalanceConfig()); + assertEquals(result.getPreCheckStatus(), PreCheckStatus.ERROR); + assertEquals(result.getMessage(), + "UNSAFE. Servers with unsafe disk utilization DURING rebalance (>=50%): " + SERVER_1 + " (52%). Enable " + + "lowDiskMode to delete segments before adding the new ones"); + } + + @Test + public void testOverThresholdOnlyDuringRebalanceIsSafeWithLowDiskMode() { + setDiskUsage(100L, 320L); + RebalancePreCheckerResult result = checkDiskUtilization(lowDiskMode()); + assertEquals(result.getPreCheckStatus(), PreCheckStatus.PASS); + assertEquals(result.getMessage(), + "Within threshold (<50%) AFTER rebalance. Servers that would go over it DURING the rebalance: " + SERVER_1 + + " (52%). lowDiskMode avoids that transient disk usage by deleting segments before adding the new ones"); + } + + @Test + public void testDowntimeCancelsLowDiskModeOut() { + // Downtime replaces the IdealState with the target assignment in one go, skipping the incremental path that is the + // only one honoring lowDiskMode, so the transient peak stands whether or not lowDiskMode is set + setDiskUsage(100L, 320L); + RebalanceConfig lowDiskModeAndDowntime = lowDiskMode(); + lowDiskModeAndDowntime.setDowntime(true); + + for (RebalanceConfig rebalanceConfig : new RebalanceConfig[]{downtime(), lowDiskModeAndDowntime}) { + RebalancePreCheckerResult result = checkDiskUtilization(rebalanceConfig); + assertEquals(result.getPreCheckStatus(), PreCheckStatus.ERROR); + assertEquals(result.getMessage(), + "UNSAFE. Servers with unsafe disk utilization DURING rebalance (>=50%): " + SERVER_1 + " (52%). lowDiskMode, " + + "which would delete segments before adding the new ones, has no effect while downtime is enabled"); + } + } + + @Test + public void testServerGainingNothingIsNotFlaggedDuringRebalance() { + // Server_0 is already at 550 of its 1000 bytes and only sheds segments. The rebalance cannot push it any higher + // and lowDiskMode would have nothing to delete first, so flagging it would blame the rebalance for a pre-existing + // condition. It drops to 350 once done, so there is nothing to report at all. + setDiskUsage(550L, 0L); + RebalancePreCheckerResult result = checkDiskUtilization(new RebalanceConfig(), SHED_ONLY_CURRENT_ASSIGNMENT, + SHED_ONLY_TARGET_ASSIGNMENT); + assertEquals(result.getPreCheckStatus(), PreCheckStatus.PASS); + assertEquals(result.getMessage(), "Within threshold (<50%)"); + } + + @Test + public void testServerGainingNothingAndStayingOverThresholdIsStillFlagged() { + // Server_0 shedding its 2 segments is not enough to bring it back under the threshold, which the AFTER estimate + // catches even though the DURING one skips it + setDiskUsage(750L, 0L); + RebalancePreCheckerResult result = checkDiskUtilization(new RebalanceConfig(), SHED_ONLY_CURRENT_ASSIGNMENT, + SHED_ONLY_TARGET_ASSIGNMENT); + assertEquals(result.getPreCheckStatus(), PreCheckStatus.ERROR); + assertEquals(result.getMessage(), + "UNSAFE. Servers with unsafe disk utilization AFTER rebalance (>=50%): " + SERVER_0 + " (55%)"); + } + + @Test + public void testDiskUsageInfoNotAvailable() { + ResourceUtilizationInfo.setDiskUsageInfo(Map.of()); + RebalancePreCheckerResult result = checkDiskUtilization(new RebalanceConfig()); + assertEquals(result.getPreCheckStatus(), PreCheckStatus.WARN); + assertTrue(result.getMessage().startsWith("Disk usage info has not been updated"), result.getMessage()); + } + + private RebalancePreCheckerResult checkDiskUtilization(RebalanceConfig rebalanceConfig) { + return checkDiskUtilization(rebalanceConfig, CURRENT_ASSIGNMENT, TARGET_ASSIGNMENT); + } + + private RebalancePreCheckerResult checkDiskUtilization(RebalanceConfig rebalanceConfig, + Map> currentAssignment, Map> targetAssignment) { + return _preChecker.checkDiskUtilization(getPreCheckContext(rebalanceConfig, currentAssignment, targetAssignment), + THRESHOLD); + } + + private static RebalanceConfig lowDiskMode() { + RebalanceConfig rebalanceConfig = new RebalanceConfig(); + rebalanceConfig.setLowDiskMode(true); + return rebalanceConfig; + } + + private static RebalanceConfig downtime() { + RebalanceConfig rebalanceConfig = new RebalanceConfig(); + rebalanceConfig.setDowntime(true); + return rebalanceConfig; + } + + private static RebalanceConfig bestEfforts() { + RebalanceConfig rebalanceConfig = new RebalanceConfig(); + rebalanceConfig.setBestEfforts(true); + return rebalanceConfig; + } + + private static void setDiskUsage(long usedSpaceBytesServer0, long usedSpaceBytesServer1) { + long now = System.currentTimeMillis(); + ResourceUtilizationInfo.setDiskUsageInfo( + Map.of(SERVER_0, new DiskUsageInfo(SERVER_0, "", TOTAL_SPACE_BYTES, usedSpaceBytesServer0, now), SERVER_1, + new DiskUsageInfo(SERVER_1, "", TOTAL_SPACE_BYTES, usedSpaceBytesServer1, now))); + } + + private static PreCheckContext getPreCheckContext(RebalanceConfig rebalanceConfig, + Map> currentAssignment, Map> targetAssignment) { + TableConfig tableConfig = new TableConfigBuilder(TableType.OFFLINE).setTableName("myTable").build(); + return new PreCheckContext("jobId", tableConfig.getTableName(), tableConfig, currentAssignment, targetAssignment, + getTableSizeDetails(), rebalanceConfig, null, null); + } + + private static TableSizeReader.TableSubTypeSizeDetails getTableSizeDetails() { + TableSizeReader.TableSubTypeSizeDetails tableSubTypeSizeDetails = new TableSizeReader.TableSubTypeSizeDetails(); + tableSubTypeSizeDetails._reportedSizePerReplicaInBytes = TABLE_SIZE_PER_REPLICA_BYTES; + return tableSubTypeSizeDetails; + } +} 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 b1f5c6e9debb..d010aa7ccb6f 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 @@ -277,11 +277,10 @@ public void testRebalance() assertEquals(rebalanceResult.getStatus(), RebalanceResult.Status.DONE); Map preCheckResult = rebalanceResult.getPreChecksResult(); assertNotNull(preCheckResult); - assertEquals(preCheckResult.size(), 6); + assertEquals(preCheckResult.size(), 5); assertTrue(preCheckResult.containsKey(DefaultRebalancePreChecker.NEEDS_RELOAD_STATUS)); assertTrue(preCheckResult.containsKey(DefaultRebalancePreChecker.IS_MINIMIZE_DATA_MOVEMENT)); - assertTrue(preCheckResult.containsKey(DefaultRebalancePreChecker.DISK_UTILIZATION_DURING_REBALANCE)); - assertTrue(preCheckResult.containsKey(DefaultRebalancePreChecker.DISK_UTILIZATION_AFTER_REBALANCE)); + assertTrue(preCheckResult.containsKey(DefaultRebalancePreChecker.DISK_UTILIZATION)); assertTrue(preCheckResult.containsKey(DefaultRebalancePreChecker.REBALANCE_CONFIG_OPTIONS)); assertTrue(preCheckResult.containsKey(DefaultRebalancePreChecker.REPLICA_GROUPS_INFO)); // Sending request to servers should fail for all, so needsPreprocess should be set to "error" to indicate that a @@ -294,14 +293,9 @@ public void testRebalance() RebalancePreCheckerResult.PreCheckStatus.PASS); assertEquals(preCheckResult.get(DefaultRebalancePreChecker.IS_MINIMIZE_DATA_MOVEMENT).getMessage(), "Instance assignment not allowed, no need for minimizeDataMovement"); - assertEquals(preCheckResult.get(DefaultRebalancePreChecker.DISK_UTILIZATION_DURING_REBALANCE).getPreCheckStatus(), + assertEquals(preCheckResult.get(DefaultRebalancePreChecker.DISK_UTILIZATION).getPreCheckStatus(), RebalancePreCheckerResult.PreCheckStatus.PASS); - assertTrue(preCheckResult.get(DefaultRebalancePreChecker.DISK_UTILIZATION_DURING_REBALANCE) - .getMessage() - .startsWith("Within threshold")); - assertEquals(preCheckResult.get(DefaultRebalancePreChecker.DISK_UTILIZATION_AFTER_REBALANCE).getPreCheckStatus(), - RebalancePreCheckerResult.PreCheckStatus.PASS); - assertTrue(preCheckResult.get(DefaultRebalancePreChecker.DISK_UTILIZATION_AFTER_REBALANCE) + assertTrue(preCheckResult.get(DefaultRebalancePreChecker.DISK_UTILIZATION) .getMessage() .startsWith("Within threshold")); assertEquals(preCheckResult.get(DefaultRebalancePreChecker.REBALANCE_CONFIG_OPTIONS).getPreCheckStatus(), @@ -1172,16 +1166,10 @@ public void testRebalancePreCheckerDiskUtil() assertEquals(rebalanceResult.getStatus(), RebalanceResult.Status.DONE); Map preCheckResult = rebalanceResult.getPreChecksResult(); assertNotNull(preCheckResult); - assertTrue(preCheckResult.containsKey(DefaultRebalancePreChecker.DISK_UTILIZATION_DURING_REBALANCE)); - assertEquals(preCheckResult.get(DefaultRebalancePreChecker.DISK_UTILIZATION_DURING_REBALANCE).getPreCheckStatus(), - RebalancePreCheckerResult.PreCheckStatus.PASS); - assertTrue(preCheckResult.get(DefaultRebalancePreChecker.DISK_UTILIZATION_DURING_REBALANCE) - .getMessage() - .startsWith("Within threshold")); - assertTrue(preCheckResult.containsKey(DefaultRebalancePreChecker.DISK_UTILIZATION_AFTER_REBALANCE)); - assertEquals(preCheckResult.get(DefaultRebalancePreChecker.DISK_UTILIZATION_AFTER_REBALANCE).getPreCheckStatus(), + assertTrue(preCheckResult.containsKey(DefaultRebalancePreChecker.DISK_UTILIZATION)); + assertEquals(preCheckResult.get(DefaultRebalancePreChecker.DISK_UTILIZATION).getPreCheckStatus(), RebalancePreCheckerResult.PreCheckStatus.PASS); - assertTrue(preCheckResult.get(DefaultRebalancePreChecker.DISK_UTILIZATION_AFTER_REBALANCE) + assertTrue(preCheckResult.get(DefaultRebalancePreChecker.DISK_UTILIZATION) .getMessage() .startsWith("Within threshold")); @@ -1199,11 +1187,24 @@ public void testRebalancePreCheckerDiskUtil() assertEquals(rebalanceResult.getStatus(), RebalanceResult.Status.DONE); preCheckResult = rebalanceResult.getPreChecksResult(); assertNotNull(preCheckResult); - assertTrue(preCheckResult.containsKey(DefaultRebalancePreChecker.DISK_UTILIZATION_DURING_REBALANCE)); - assertEquals(preCheckResult.get(DefaultRebalancePreChecker.DISK_UTILIZATION_DURING_REBALANCE).getPreCheckStatus(), + assertTrue(preCheckResult.containsKey(DefaultRebalancePreChecker.DISK_UTILIZATION)); + assertEquals(preCheckResult.get(DefaultRebalancePreChecker.DISK_UTILIZATION).getPreCheckStatus(), RebalancePreCheckerResult.PreCheckStatus.ERROR); - assertTrue(preCheckResult.containsKey(DefaultRebalancePreChecker.DISK_UTILIZATION_AFTER_REBALANCE)); - assertEquals(preCheckResult.get(DefaultRebalancePreChecker.DISK_UTILIZATION_AFTER_REBALANCE).getPreCheckStatus(), + assertTrue(preCheckResult.get(DefaultRebalancePreChecker.DISK_UTILIZATION) + .getMessage() + .startsWith("UNSAFE. Servers with unsafe disk utilization AFTER rebalance")); + + // The servers are over the threshold no matter how the rebalance is run, so lowDiskMode does not help + rebalanceConfig = new RebalanceConfig(); + rebalanceConfig.setDryRun(true); + rebalanceConfig.setPreChecks(true); + rebalanceConfig.setLowDiskMode(true); + + rebalanceResult = tableRebalancer.rebalance(tableConfig, rebalanceConfig, null); + assertEquals(rebalanceResult.getStatus(), RebalanceResult.Status.DONE); + preCheckResult = rebalanceResult.getPreChecksResult(); + assertNotNull(preCheckResult); + assertEquals(preCheckResult.get(DefaultRebalancePreChecker.DISK_UTILIZATION).getPreCheckStatus(), RebalancePreCheckerResult.PreCheckStatus.ERROR); _helixResourceManager.deleteOfflineTable(RAW_TABLE_NAME); @@ -1317,6 +1318,17 @@ public void testRebalancePreCheckerRebalanceConfig() assertEquals(preCheckerResult.getPreCheckStatus(), RebalancePreCheckerResult.PreCheckStatus.PASS); assertEquals(preCheckerResult.getMessage(), "All rebalance parameters look good"); + // trigger the warning about lowDiskMode being inert with downtime, which replaces the IdealState in one go + rebalanceConfig.setLowDiskMode(true); + rebalanceResult = tableRebalancer.rebalance(newTableConfig, rebalanceConfig, null); + preCheckerResult = rebalanceResult.getPreChecksResult().get(DefaultRebalancePreChecker.REBALANCE_CONFIG_OPTIONS); + assertNotNull(preCheckerResult); + assertEquals(preCheckerResult.getPreCheckStatus(), RebalancePreCheckerResult.PreCheckStatus.WARN); + assertEquals(preCheckerResult.getMessage(), + "lowDiskMode has no effect when downtime is enabled, disable downtime for segments to be deleted before the " + + "new ones are added."); + rebalanceConfig.setLowDiskMode(false); + // trigger peer-download enabled table rebalance warning newTableConfig.getValidationConfig().setPeerSegmentDownloadScheme("http"); diff --git a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/TableRebalanceIntegrationTest.java b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/TableRebalanceIntegrationTest.java index 30eceaba1c3f..d7e7713aaf22 100644 --- a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/TableRebalanceIntegrationTest.java +++ b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/TableRebalanceIntegrationTest.java @@ -931,11 +931,10 @@ private void checkRebalancePreCheckStatus(RebalanceResult rebalanceResult, Rebal assertEquals(rebalanceResult.getStatus(), expectedStatus); Map preChecksResult = rebalanceResult.getPreChecksResult(); assertNotNull(preChecksResult); - assertEquals(preChecksResult.size(), 6); + assertEquals(preChecksResult.size(), 5); assertTrue(preChecksResult.containsKey(DefaultRebalancePreChecker.IS_MINIMIZE_DATA_MOVEMENT)); assertTrue(preChecksResult.containsKey(DefaultRebalancePreChecker.NEEDS_RELOAD_STATUS)); - assertTrue(preChecksResult.containsKey(DefaultRebalancePreChecker.DISK_UTILIZATION_DURING_REBALANCE)); - assertTrue(preChecksResult.containsKey(DefaultRebalancePreChecker.DISK_UTILIZATION_AFTER_REBALANCE)); + assertTrue(preChecksResult.containsKey(DefaultRebalancePreChecker.DISK_UTILIZATION)); assertTrue(preChecksResult.containsKey(DefaultRebalancePreChecker.REBALANCE_CONFIG_OPTIONS)); assertEquals(preChecksResult.get(DefaultRebalancePreChecker.IS_MINIMIZE_DATA_MOVEMENT).getPreCheckStatus(), expectedMinimizeDataMovementStatus); @@ -957,9 +956,7 @@ private void checkRebalancePreCheckStatus(RebalanceResult rebalanceResult, Rebal // .RESOURCE_UTILIZATION_CHECKER_INITIAL_DELAY was set to 30000s, see org.apache.pinot.controller.helix // .ControllerTest.getDefaultControllerConfiguration), server's disk util should be unavailable on all servers if // not explicitly set via org.apache.pinot.controller.validation.ResourceUtilizationInfo.setDiskUsageInfo - assertEquals(preChecksResult.get(DefaultRebalancePreChecker.DISK_UTILIZATION_DURING_REBALANCE).getPreCheckStatus(), - RebalancePreCheckerResult.PreCheckStatus.WARN); - assertEquals(preChecksResult.get(DefaultRebalancePreChecker.DISK_UTILIZATION_AFTER_REBALANCE).getPreCheckStatus(), + assertEquals(preChecksResult.get(DefaultRebalancePreChecker.DISK_UTILIZATION).getPreCheckStatus(), RebalancePreCheckerResult.PreCheckStatus.WARN); }