Skip to content

Commit

Permalink
add RebalanceConfig class to replace use of BaseConfiguration for reb…
Browse files Browse the repository at this point in the history 
…alance configs (#11730)

* add RebalanceConfig class to replace use of Configuration for rebalance configs

* remove RebalanceConfigConstants
  • Loading branch information
@klsince
klsince authored Oct 4, 2023
1 parent 7ed9aa3 commit c96221f
Show file tree
Hide file tree
Showing 30 changed files with 440 additions and 509 deletions.
6 changes: 3 additions & 3 deletions pinot-controller/src/main/java/org/apache/pinot/controller/ControllerConf.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -30,10 +30,10 @@
import org.apache.commons.lang3.StringUtils;
import org.apache.helix.controller.rebalancer.strategy.AutoRebalanceStrategy;
import org.apache.pinot.common.protocols.SegmentCompletionProtocol;
import org.apache.pinot.controller.helix.core.rebalance.RebalanceConfig;
import org.apache.pinot.spi.env.PinotConfiguration;
import org.apache.pinot.spi.filesystem.LocalPinotFS;
import org.apache.pinot.spi.utils.CommonConstants;
import org.apache.pinot.spi.utils.RebalanceConfigConstants;
import org.apache.pinot.spi.utils.TimeUtils;

import static org.apache.pinot.spi.utils.CommonConstants.Controller.CONFIG_OF_CONTROLLER_METRICS_PREFIX;
Expand Down Expand Up @@ -962,12 +962,12 @@ public boolean enableSegmentRelocatorLocalTierMigration() {

public long getSegmentRelocatorExternalViewCheckIntervalInMs() {
return getProperty(ControllerPeriodicTasksConf.SEGMENT_RELOCATOR_EXTERNAL_VIEW_CHECK_INTERVAL_IN_MS,
RebalanceConfigConstants.DEFAULT_EXTERNAL_VIEW_CHECK_INTERVAL_IN_MS);
RebalanceConfig.DEFAULT_EXTERNAL_VIEW_CHECK_INTERVAL_IN_MS);
}

public long getSegmentRelocatorExternalViewStabilizationTimeoutInMs() {
return getProperty(ControllerPeriodicTasksConf.SEGMENT_RELOCATOR_EXTERNAL_VIEW_STABILIZATION_TIMEOUT_IN_MS,
RebalanceConfigConstants.DEFAULT_EXTERNAL_VIEW_STABILIZATION_TIMEOUT_IN_MS);
RebalanceConfig.DEFAULT_EXTERNAL_VIEW_STABILIZATION_TIMEOUT_IN_MS);
}

public boolean isSegmentRelocatorRebalanceTablesSequentially() {
Expand Down
46 changes: 21 additions & 25 deletions ...er/src/main/java/org/apache/pinot/controller/api/resources/PinotTableRestletResource.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -64,8 +64,6 @@
import javax.ws.rs.core.HttpHeaders;
import javax.ws.rs.core.MediaType;
import javax.ws.rs.core.Response;
import org.apache.commons.configuration.BaseConfiguration;
import org.apache.commons.configuration.Configuration;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.tuple.Pair;
import org.apache.helix.AccessOption;
Expand Down Expand Up @@ -94,6 +92,8 @@
import org.apache.pinot.controller.helix.core.PinotHelixResourceManager;
import org.apache.pinot.controller.helix.core.PinotResourceManagerResponse;
import org.apache.pinot.controller.helix.core.minion.PinotHelixTaskResourceManager;
import org.apache.pinot.controller.helix.core.rebalance.RebalanceConfig;
import org.apache.pinot.controller.helix.core.rebalance.RebalanceJobConstants;
import org.apache.pinot.controller.helix.core.rebalance.RebalanceResult;
import org.apache.pinot.controller.helix.core.rebalance.TableRebalanceProgressStats;
import org.apache.pinot.controller.helix.core.rebalance.TableRebalancer;
Expand All @@ -114,7 +114,6 @@
import org.apache.pinot.spi.data.Schema;
import org.apache.pinot.spi.utils.CommonConstants;
import org.apache.pinot.spi.utils.JsonUtils;
import org.apache.pinot.spi.utils.RebalanceConfigConstants;
import org.apache.pinot.spi.utils.builder.TableNameBuilder;
import org.apache.pinot.spi.utils.retry.RetryPolicies;
import org.apache.zookeeper.data.Stat;
Expand Down Expand Up @@ -621,34 +620,32 @@ public RebalanceResult rebalance(

String tableNameWithType = constructTableNameWithType(tableName, tableTypeStr);

Configuration rebalanceConfig = new BaseConfiguration();
rebalanceConfig.addProperty(RebalanceConfigConstants.DRY_RUN, dryRun);
rebalanceConfig.addProperty(RebalanceConfigConstants.REASSIGN_INSTANCES, reassignInstances);
rebalanceConfig.addProperty(RebalanceConfigConstants.INCLUDE_CONSUMING, includeConsuming);
rebalanceConfig.addProperty(RebalanceConfigConstants.BOOTSTRAP, bootstrap);
rebalanceConfig.addProperty(RebalanceConfigConstants.DOWNTIME, downtime);
rebalanceConfig.addProperty(RebalanceConfigConstants.MIN_REPLICAS_TO_KEEP_UP_FOR_NO_DOWNTIME, minAvailableReplicas);
rebalanceConfig.addProperty(RebalanceConfigConstants.BEST_EFFORTS, bestEfforts);
rebalanceConfig.addProperty(RebalanceConfigConstants.EXTERNAL_VIEW_CHECK_INTERVAL_IN_MS,
externalViewCheckIntervalInMs);
rebalanceConfig.addProperty(RebalanceConfigConstants.EXTERNAL_VIEW_STABILIZATION_TIMEOUT_IN_MS,
externalViewStabilizationTimeoutInMs);
rebalanceConfig.addProperty(RebalanceConfigConstants.UPDATE_TARGET_TIER, updateTargetTier);
rebalanceConfig.addProperty(RebalanceConfigConstants.JOB_ID, TableRebalancer.createUniqueRebalanceJobIdentifier());
RebalanceConfig rebalanceConfig = new RebalanceConfig();
rebalanceConfig.setDryRun(dryRun);
rebalanceConfig.setReassignInstances(reassignInstances);
rebalanceConfig.setIncludeConsuming(includeConsuming);
rebalanceConfig.setBootstrap(bootstrap);
rebalanceConfig.setDowntime(downtime);
rebalanceConfig.setMinAvailableReplicas(minAvailableReplicas);
rebalanceConfig.setBestEfforts(bestEfforts);
rebalanceConfig.setExternalViewCheckIntervalInMs(externalViewCheckIntervalInMs);
rebalanceConfig.setExternalViewStabilizationTimeoutInMs(externalViewStabilizationTimeoutInMs);
rebalanceConfig.setUpdateTargetTier(updateTargetTier);
rebalanceConfig.setJobId(TableRebalancer.createUniqueRebalanceJobIdentifier());

try {
if (dryRun || downtime) {
// For dry-run or rebalance with downtime, directly return the rebalance result as it should return immediately
return _pinotHelixResourceManager.rebalanceTable(tableNameWithType, rebalanceConfig, false);
} else {
// Make a dry-run first to get the target assignment
rebalanceConfig.setProperty(RebalanceConfigConstants.DRY_RUN, true);
rebalanceConfig.setDryRun(true);
RebalanceResult dryRunResult =
_pinotHelixResourceManager.rebalanceTable(tableNameWithType, rebalanceConfig, false);

if (dryRunResult.getStatus() == RebalanceResult.Status.DONE) {
// If dry-run succeeded, run rebalance asynchronously
rebalanceConfig.setProperty(RebalanceConfigConstants.DRY_RUN, false);
rebalanceConfig.setDryRun(false);
_executorService.submit(() -> {
try {
_pinotHelixResourceManager.rebalanceTable(tableNameWithType, rebalanceConfig, true);
Expand Down Expand Up @@ -744,13 +741,12 @@ public ServerRebalanceJobStatusResponse rebalanceStatus(
throw new ControllerApplicationException(LOGGER, "Failed to find controller job id: " + jobId,
Response.Status.NOT_FOUND);
}
TableRebalanceProgressStats tableRebalanceProgressStats =
JsonUtils.stringToObject(controllerJobZKMetadata.get(RebalanceConfigConstants.REBALANCE_PROGRESS_STATS),
TableRebalanceProgressStats.class);
TableRebalanceProgressStats tableRebalanceProgressStats = JsonUtils.stringToObject(
controllerJobZKMetadata.get(RebalanceJobConstants.JOB_METADATA_KEY_REBALANCE_PROGRESS_STATS),
TableRebalanceProgressStats.class);
long timeSinceStartInSecs = 0L;
if (!tableRebalanceProgressStats.getStatus().equals(RebalanceResult.Status.DONE)) {
timeSinceStartInSecs =
(System.currentTimeMillis() - tableRebalanceProgressStats.getStartTimeMs()) / 1000;
if (!RebalanceResult.Status.DONE.toString().equals(tableRebalanceProgressStats.getStatus())) {
timeSinceStartInSecs = (System.currentTimeMillis() - tableRebalanceProgressStats.getStartTimeMs()) / 1000;
}

ServerRebalanceJobStatusResponse serverRebalanceJobStatusResponse = new ServerRebalanceJobStatusResponse();
Expand Down
16 changes: 8 additions & 8 deletions ...r/src/main/java/org/apache/pinot/controller/api/resources/PinotTenantRestletResource.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -58,7 +58,8 @@
import org.apache.pinot.controller.api.exception.ControllerApplicationException;
import org.apache.pinot.controller.helix.core.PinotHelixResourceManager;
import org.apache.pinot.controller.helix.core.PinotResourceManagerResponse;
import org.apache.pinot.controller.helix.core.rebalance.tenant.TenantRebalanceContext;
import org.apache.pinot.controller.helix.core.rebalance.RebalanceJobConstants;
import org.apache.pinot.controller.helix.core.rebalance.tenant.TenantRebalanceConfig;
import org.apache.pinot.controller.helix.core.rebalance.tenant.TenantRebalanceProgressStats;
import org.apache.pinot.controller.helix.core.rebalance.tenant.TenantRebalanceResult;
import org.apache.pinot.controller.helix.core.rebalance.tenant.TenantRebalancer;
Expand All @@ -70,7 +71,6 @@
import org.apache.pinot.spi.config.tenant.Tenant;
import org.apache.pinot.spi.config.tenant.TenantRole;
import org.apache.pinot.spi.utils.JsonUtils;
import org.apache.pinot.spi.utils.RebalanceConfigConstants;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

Expand Down Expand Up @@ -584,9 +584,9 @@ public SuccessResponse deleteTenant(
@ApiOperation(value = "Rebalances all the tables that are part of the tenant")
public TenantRebalanceResult rebalance(
@ApiParam(value = "Name of the tenant whose table are to be rebalanced", required = true)
@PathParam("tenantName") String tenantName, @ApiParam(required = true) TenantRebalanceContext context) {
context.setTenantName(tenantName);
return _tenantRebalancer.rebalance(context);
@PathParam("tenantName") String tenantName, @ApiParam(required = true) TenantRebalanceConfig config) {
config.setTenantName(tenantName);
return _tenantRebalancer.rebalance(config);
}

@GET
Expand All @@ -606,9 +606,9 @@ public TenantRebalanceJobStatusResponse rebalanceStatus(
throw new ControllerApplicationException(LOGGER, "Failed to find controller job id: " + jobId,
Response.Status.NOT_FOUND);
}
TenantRebalanceProgressStats tenantRebalanceProgressStats =
JsonUtils.stringToObject(controllerJobZKMetadata.get(RebalanceConfigConstants.REBALANCE_PROGRESS_STATS),
TenantRebalanceProgressStats.class);
TenantRebalanceProgressStats tenantRebalanceProgressStats = JsonUtils.stringToObject(
controllerJobZKMetadata.get(RebalanceJobConstants.JOB_METADATA_KEY_REBALANCE_PROGRESS_STATS),
TenantRebalanceProgressStats.class);
long timeSinceStartInSecs = tenantRebalanceProgressStats.getTimeToFinishInSeconds();
if (tenantRebalanceProgressStats.getCompletionStatusMsg() == null) {
timeSinceStartInSecs =
Expand Down
14 changes: 6 additions & 8 deletions ...oller/src/main/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManager.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -56,7 +56,6 @@
import javax.ws.rs.NotFoundException;
import javax.ws.rs.core.Response;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.configuration.Configuration;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.tuple.Pair;
import org.apache.helix.AccessOption;
Expand Down Expand Up @@ -144,6 +143,7 @@
import org.apache.pinot.controller.helix.core.lineage.LineageManager;
import org.apache.pinot.controller.helix.core.lineage.LineageManagerFactory;
import org.apache.pinot.controller.helix.core.realtime.PinotLLCRealtimeSegmentManager;
import org.apache.pinot.controller.helix.core.rebalance.RebalanceConfig;
import org.apache.pinot.controller.helix.core.rebalance.RebalanceResult;
import org.apache.pinot.controller.helix.core.rebalance.TableRebalancer;
import org.apache.pinot.controller.helix.core.rebalance.ZkBasedTableRebalanceObserver;
Expand Down Expand Up @@ -175,7 +175,6 @@
import org.apache.pinot.spi.utils.IngestionConfigUtils;
import org.apache.pinot.spi.utils.InstanceTypeUtils;
import org.apache.pinot.spi.utils.JsonUtils;
import org.apache.pinot.spi.utils.RebalanceConfigConstants;
import org.apache.pinot.spi.utils.TimeUtils;
import org.apache.pinot.spi.utils.builder.TableNameBuilder;
import org.apache.pinot.spi.utils.retry.RetryPolicies;
Expand Down Expand Up @@ -3081,21 +3080,20 @@ private PinotResourceManagerResponse enableInstance(String instanceName, boolean
* Entry point for table Rebalacing.
* @param tableNameWithType
* @param rebalanceConfig
* @param trackRebalanceProgress - Do we want to track rebalance progress stats
* @param trackRebalanceProgress whether to track rebalance progress stats
* @return RebalanceResult
* @throws TableNotFoundException
*/
public RebalanceResult rebalanceTable(String tableNameWithType, Configuration rebalanceConfig,
public RebalanceResult rebalanceTable(String tableNameWithType, RebalanceConfig rebalanceConfig,
boolean trackRebalanceProgress)
throws TableNotFoundException {
TableConfig tableConfig = getTableConfig(tableNameWithType);
if (tableConfig == null) {
throw new TableNotFoundException("Failed to find table config for table: " + tableNameWithType);
}
String rebalanceJobId = rebalanceConfig.getString(RebalanceConfigConstants.JOB_ID);
Preconditions.checkState(rebalanceJobId != null, "RebalanceId not populated in the rebalanceConfig ");
if (rebalanceConfig.getBoolean(RebalanceConfigConstants.UPDATE_TARGET_TIER,
RebalanceConfigConstants.DEFAULT_UPDATE_TARGET_TIER)) {
String rebalanceJobId = rebalanceConfig.getJobId();
Preconditions.checkState(rebalanceJobId != null, "RebalanceId not populated in the rebalanceConfig");
if (rebalanceConfig.isUpdateTargetTier()) {
updateTargetTier(rebalanceJobId, tableNameWithType, tableConfig);
}
ZkBasedTableRebalanceObserver zkBasedTableRebalanceObserver = null;
Expand Down
9 changes: 3 additions & 6 deletions ...a/org/apache/pinot/controller/helix/core/assignment/segment/OfflineSegmentAssignment.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -23,15 +23,14 @@
import java.util.Map;
import javax.annotation.Nullable;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.configuration.Configuration;
import org.apache.commons.lang3.tuple.Pair;
import org.apache.pinot.common.assignment.InstancePartitions;
import org.apache.pinot.common.tier.Tier;
import org.apache.pinot.controller.helix.core.assignment.segment.strategy.AllServersSegmentAssignmentStrategy;
import org.apache.pinot.controller.helix.core.assignment.segment.strategy.SegmentAssignmentStrategy;
import org.apache.pinot.controller.helix.core.assignment.segment.strategy.SegmentAssignmentStrategyFactory;
import org.apache.pinot.controller.helix.core.rebalance.RebalanceConfig;
import org.apache.pinot.spi.config.table.assignment.InstancePartitionsType;
import org.apache.pinot.spi.utils.RebalanceConfigConstants;


/**
Expand Down Expand Up @@ -61,7 +60,7 @@ public List<String> assignSegment(String segmentName, Map<String, Map<String, St
@Override
public Map<String, Map<String, String>> rebalanceTable(Map<String, Map<String, String>> currentAssignment,
Map<InstancePartitionsType, InstancePartitions> instancePartitionsMap, @Nullable List<Tier> sortedTiers,
@Nullable Map<String, InstancePartitions> tierInstancePartitionsMap, Configuration config) {
@Nullable Map<String, InstancePartitions> tierInstancePartitionsMap, RebalanceConfig config) {
InstancePartitions offlineInstancePartitions = instancePartitionsMap.get(InstancePartitionsType.OFFLINE);
Preconditions
.checkState(offlineInstancePartitions != null, "Failed to find OFFLINE instance partitions for table: %s",
Expand All @@ -78,9 +77,7 @@ public Map<String, Map<String, String>> rebalanceTable(Map<String, Map<String, S
return segmentAssignmentStrategy
.reassignSegments(currentAssignment, offlineInstancePartitions, InstancePartitionsType.OFFLINE);
}
boolean bootstrap =
config.getBoolean(RebalanceConfigConstants.BOOTSTRAP, RebalanceConfigConstants.DEFAULT_BOOTSTRAP);

boolean bootstrap = config.isBootstrap();
// Rebalance tiers first
Pair<List<Map<String, Map<String, String>>>, Map<String, Map<String, String>>> pair =
rebalanceTiers(currentAssignment, sortedTiers, tierInstancePartitionsMap, bootstrap,
Expand Down
12 changes: 4 additions & 8 deletions .../org/apache/pinot/controller/helix/core/assignment/segment/RealtimeSegmentAssignment.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -25,15 +25,14 @@
import java.util.TreeMap;
import javax.annotation.Nullable;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.configuration.Configuration;
import org.apache.commons.lang3.tuple.Pair;
import org.apache.pinot.common.assignment.InstancePartitions;
import org.apache.pinot.common.tier.Tier;
import org.apache.pinot.controller.helix.core.assignment.segment.strategy.SegmentAssignmentStrategy;
import org.apache.pinot.controller.helix.core.assignment.segment.strategy.SegmentAssignmentStrategyFactory;
import org.apache.pinot.controller.helix.core.rebalance.RebalanceConfig;
import org.apache.pinot.spi.config.table.assignment.InstancePartitionsType;
import org.apache.pinot.spi.utils.CommonConstants.Helix.StateModel.SegmentStateModel;
import org.apache.pinot.spi.utils.RebalanceConfigConstants;


/**
Expand Down Expand Up @@ -174,17 +173,14 @@ protected List<String> assignConsumingSegment(int segmentPartitionId, InstancePa
@Override
public Map<String, Map<String, String>> rebalanceTable(Map<String, Map<String, String>> currentAssignment,
Map<InstancePartitionsType, InstancePartitions> instancePartitionsMap, @Nullable List<Tier> sortedTiers,
@Nullable Map<String, InstancePartitions> tierInstancePartitionsMap, Configuration config) {
@Nullable Map<String, InstancePartitions> tierInstancePartitionsMap, RebalanceConfig config) {
InstancePartitions completedInstancePartitions = instancePartitionsMap.get(InstancePartitionsType.COMPLETED);
InstancePartitions consumingInstancePartitions = instancePartitionsMap.get(InstancePartitionsType.CONSUMING);
Preconditions
.checkState(consumingInstancePartitions != null, "Failed to find CONSUMING instance partitions for table: %s",
_tableNameWithType);
boolean includeConsuming = config
.getBoolean(RebalanceConfigConstants.INCLUDE_CONSUMING, RebalanceConfigConstants.DEFAULT_INCLUDE_CONSUMING);
boolean bootstrap =
config.getBoolean(RebalanceConfigConstants.BOOTSTRAP, RebalanceConfigConstants.DEFAULT_BOOTSTRAP);

boolean includeConsuming = config.isIncludeConsuming();
boolean bootstrap = config.isBootstrap();
// Rebalance tiers first
Pair<List<Map<String, Map<String, String>>>, Map<String, Map<String, String>>> pair =
rebalanceTiers(currentAssignment, sortedTiers, tierInstancePartitionsMap, bootstrap,
Expand Down
4 changes: 2 additions & 2 deletions ...ain/java/org/apache/pinot/controller/helix/core/assignment/segment/SegmentAssignment.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -21,11 +21,11 @@
import java.util.List;
import java.util.Map;
import javax.annotation.Nullable;
import org.apache.commons.configuration.Configuration;
import org.apache.helix.HelixManager;
import org.apache.pinot.common.assignment.InstancePartitions;
import org.apache.pinot.common.metrics.ControllerMetrics;
import org.apache.pinot.common.tier.Tier;
import org.apache.pinot.controller.helix.core.rebalance.RebalanceConfig;
import org.apache.pinot.spi.config.table.TableConfig;
import org.apache.pinot.spi.config.table.assignment.InstancePartitionsType;

Expand Down Expand Up @@ -65,5 +65,5 @@ List<String> assignSegment(String segmentName, Map<String, Map<String, String>>
*/
Map<String, Map<String, String>> rebalanceTable(Map<String, Map<String, String>> currentAssignment,
Map<InstancePartitionsType, InstancePartitions> instancePartitionsMap, @Nullable List<Tier> sortedTiers,
@Nullable Map<String, InstancePartitions> tierInstancePartitionsMap, Configuration config);
@Nullable Map<String, InstancePartitions> tierInstancePartitionsMap, RebalanceConfig config);
}
Loading

0 comments on commit c96221f

Please sign in to comment.