You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pinot.apache.org by "klsince (via GitHub)" <gi...@apache.org> on 2023/10/04 21:34:34 UTC
[PR] Add support to retry failed table rebalance [pinot]
klsince opened a new pull request, #11740:
URL: https://github.com/apache/pinot/pull/11740
Currently, table rebalance triggered by user runs at best effort. It could fail if the controller running it got restarted; or some servers were not stable, making the rebalance timed out while waiting for external view to converge with ideal state, etc.
This PR adds support to retry failed table rebalance:
1. extended rebalance job status updating mechanism to help detect failed rebalance. The rebalance job is considered as failed if it's in FAILED status or in IN_PROGRESS but didn't send out heartbeat for too long;
2. added a controller periodic task - RebalanceChecker - to detect failed rebalance jobs and do retry.
3. added configs to disable checker, enable it to emit failure metrics only, or enable it and allow it to kick off retry
## Release note ##
New configs for the RebalanceChecker periodic task:
1. controller.rebalance.checker.frequencyPeriod: 5min by default, with -1 to disable it
2. controller.rebalanceChecker.initialDelayInSeconds: 2min+ by default
3. controller.rebalanceChecker.checkOnly: true by default, so just check and emit failure metrics
New configs added for RebalanceConfig:
1. heartbeatIntervalInMs: 300_000 i.e. 5min
2. heartbeatTimeoutInMs: 3600_000 i.e. 1hr
3. maxRetry: 3 by default
New metrics to monitor rebalance and its retries:
1. TABLE_REBALANCE_FAILURE("TableRebalanceFailure", false), emit from TableRebalancer.rebalanceTable()
2. TABLE_REBALANCE_EXECUTION_TIME_MS("tableRebalanceExecutionTimeMs", false), emit from TableRebalancer.rebalanceTable()
3. TABLE_REBALANCE_FAILURE_DETECTED("TableRebalanceFailureDetected", false), emit from RebalanceChecker
4. TABLE_REBALANCE_RETRY("TableRebalanceRetry", false), emit from RebalanceChecker
--
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
Re: [PR] Add support to retry failed table rebalance [pinot]
Posted by "klsince (via GitHub)" <gi...@apache.org>.
klsince commented on code in PR #11740:
URL: https://github.com/apache/pinot/pull/11740#discussion_r1355830231
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/rebalance/RebalanceChecker.java:
##########
@@ -0,0 +1,279 @@
+/**
+ * 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 com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Preconditions;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ExecutorService;
+import org.apache.commons.lang3.RandomUtils;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.commons.lang3.tuple.Pair;
+import org.apache.pinot.common.metadata.ZKMetadataProvider;
+import org.apache.pinot.common.metadata.controllerjob.ControllerJobType;
+import org.apache.pinot.common.metrics.ControllerMeter;
+import org.apache.pinot.common.metrics.ControllerMetrics;
+import org.apache.pinot.controller.ControllerConf;
+import org.apache.pinot.controller.LeadControllerManager;
+import org.apache.pinot.controller.helix.core.PinotHelixResourceManager;
+import org.apache.pinot.controller.helix.core.periodictask.ControllerPeriodicTask;
+import org.apache.pinot.spi.config.table.TableConfig;
+import org.apache.pinot.spi.utils.CommonConstants;
+import org.apache.pinot.spi.utils.JsonUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+/**
+ * Periodic task to check whether a user triggered rebalance job is completed or not, and retry if failed. The retry
+ * job is started with the same rebalance configs provided by the user and does best effort to stop the other jobs
+ * for the same table. This task can be configured to just check failures and report metrics, and not to do retry.
+ */
+public class RebalanceChecker extends ControllerPeriodicTask<Void> {
+ private static final Logger LOGGER = LoggerFactory.getLogger(RebalanceChecker.class);
+ private static final double RETRY_DELAY_SCALE_FACTOR = 2.0;
+ private final boolean _checkOnly;
+ private final ExecutorService _executorService;
+
+ public RebalanceChecker(PinotHelixResourceManager pinotHelixResourceManager,
+ LeadControllerManager leadControllerManager, ControllerConf config, ControllerMetrics controllerMetrics,
+ ExecutorService executorService) {
+ super(RebalanceChecker.class.getSimpleName(), config.getRebalanceCheckerFrequencyInSeconds(),
+ config.getRebalanceCheckerInitialDelayInSeconds(), pinotHelixResourceManager, leadControllerManager,
+ controllerMetrics);
+ _checkOnly = config.isRebalanceCheckerCheckOnly();
+ _executorService = executorService;
+ }
+
+ @Override
+ protected void processTable(String tableNameWithType) {
+ _executorService.submit(() -> {
+ try {
+ LOGGER.info("Start to retry rebalance for table: {}", tableNameWithType);
+ TableConfig tableConfig = _pinotHelixResourceManager.getTableConfig(tableNameWithType);
+ Preconditions.checkState(tableConfig != null, "Failed to find table config for table: {}", tableNameWithType);
+ retryRebalanceTable(tableNameWithType, tableConfig);
+ } catch (Throwable t) {
+ LOGGER.error("Failed to retry rebalance for table: {}", tableNameWithType, t);
+ }
+ });
+ }
+
+ @VisibleForTesting
+ void retryRebalanceTable(String tableNameWithType, TableConfig tableConfig)
+ throws Exception {
+ // Get all rebalance jobs as tracked in ZK for this table to check if there is any failure to retry.
+ Map<String, Map<String, String>> allJobMetadata = _pinotHelixResourceManager.getAllJobsForTable(tableNameWithType,
Review Comment:
All tables' rebalance job status is kept in one ZNode, and there is cache for zkclient.get(), so I think it'd be fine to process tables separately. But good point that `_executorService.submit` shouldn't be called immediately in processTable() but when a retry is really needed for the table.
--
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
Re: [PR] Add support to retry failed table rebalance [pinot]
Posted by "klsince (via GitHub)" <gi...@apache.org>.
klsince commented on code in PR #11740:
URL: https://github.com/apache/pinot/pull/11740#discussion_r1361254945
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManager.java:
##########
@@ -2063,28 +2070,76 @@ public boolean addNewReloadAllSegmentsJob(String tableNameWithType, String jobId
ZKMetadataProvider.constructPropertyStorePathForControllerJob(ControllerJobType.RELOAD_SEGMENT));
}
+ /**
+ * Adds a new job metadata for controller job like table rebalance or reload into ZK
+ * @param jobId job's UUID
+ * @param jobMetadata the job metadata
+ * @param jobResourcePath zkPath to add the new job metadata
+ * @return boolean representing success / failure of the ZK write step
+ */
public boolean addControllerJobToZK(String jobId, Map<String, String> jobMetadata, String jobResourcePath) {
+ return addControllerJobToZK(jobId, jobMetadata, jobResourcePath, prev -> true);
+ }
+
+ /**
+ * Adds a new job metadata for controller job like table rebalance or reload into ZK
+ * @param jobId job's UUID
+ * @param jobMetadata the job metadata
+ * @param jobResourcePath zkPath to add the new job metadata
+ * @param prevJobMetadataChecker to check the previous job metadata before adding new one
+ * @return boolean representing success / failure of the ZK write step
+ */
+ public boolean addControllerJobToZK(String jobId, Map<String, String> jobMetadata, String jobResourcePath,
+ Predicate<Map<String, String>> prevJobMetadataChecker) {
Preconditions.checkState(jobMetadata.get(CommonConstants.ControllerJob.SUBMISSION_TIME_MS) != null,
"Submission Time in JobMetadata record not set. Cannot expire these records");
Stat stat = new Stat();
- ZNRecord tableJobsZnRecord = _propertyStore.get(jobResourcePath, stat, AccessOption.PERSISTENT);
- if (tableJobsZnRecord != null) {
- Map<String, Map<String, String>> tasks = tableJobsZnRecord.getMapFields();
- tasks.put(jobId, jobMetadata);
- if (tasks.size() > CommonConstants.ControllerJob.MAXIMUM_CONTROLLER_JOBS_IN_ZK) {
- tasks = tasks.entrySet().stream().sorted((v1, v2) -> Long.compare(
+ ZNRecord jobsZnRecord = _propertyStore.get(jobResourcePath, stat, AccessOption.PERSISTENT);
+ if (jobsZnRecord != null) {
+ Map<String, Map<String, String>> jobMetadataMap = jobsZnRecord.getMapFields();
+ Map<String, String> prevJobMetadata = jobMetadataMap.get(jobId);
+ if (!prevJobMetadataChecker.test(prevJobMetadata)) {
+ return false;
+ }
+ jobMetadataMap.put(jobId, jobMetadata);
+ if (jobMetadataMap.size() > CommonConstants.ControllerJob.MAXIMUM_CONTROLLER_JOBS_IN_ZK) {
+ jobMetadataMap = jobMetadataMap.entrySet().stream().sorted((v1, v2) -> Long.compare(
Long.parseLong(v2.getValue().get(CommonConstants.ControllerJob.SUBMISSION_TIME_MS)),
Long.parseLong(v1.getValue().get(CommonConstants.ControllerJob.SUBMISSION_TIME_MS))))
.collect(Collectors.toList()).subList(0, CommonConstants.ControllerJob.MAXIMUM_CONTROLLER_JOBS_IN_ZK)
.stream().collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
}
- tableJobsZnRecord.setMapFields(tasks);
- return _propertyStore.set(jobResourcePath, tableJobsZnRecord, stat.getVersion(), AccessOption.PERSISTENT);
+ jobsZnRecord.setMapFields(jobMetadataMap);
+ return _propertyStore.set(jobResourcePath, jobsZnRecord, stat.getVersion(), AccessOption.PERSISTENT);
} else {
- tableJobsZnRecord = new ZNRecord(jobResourcePath);
- tableJobsZnRecord.setMapField(jobId, jobMetadata);
- return _propertyStore.set(jobResourcePath, tableJobsZnRecord, AccessOption.PERSISTENT);
+ jobsZnRecord = new ZNRecord(jobResourcePath);
+ jobsZnRecord.setMapField(jobId, jobMetadata);
+ return _propertyStore.set(jobResourcePath, jobsZnRecord, AccessOption.PERSISTENT);
+ }
+ }
+
+ /**
+ * Update existing job metadata belong to the table
+ * @param tableNameWithType whose job metadata to be updated
+ * @param jobResourcePath zkPath to look for the job metadata for the table
+ * @param updater to modify the job metadata in place
+ * @return boolean representing success / failure of the ZK write step
+ */
+ public boolean updateAllJobsForTable(String tableNameWithType, String jobResourcePath,
Review Comment:
sgtm
> making the updater on the whole jobMetadataMap which is more general
I didn't get this one. the updater is to handle the whole `jobMetadataMap` currently.
--
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
Re: [PR] Add support to retry failed table rebalance [pinot]
Posted by "Jackie-Jiang (via GitHub)" <gi...@apache.org>.
Jackie-Jiang commented on code in PR #11740:
URL: https://github.com/apache/pinot/pull/11740#discussion_r1363095650
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/rebalance/RebalanceChecker.java:
##########
@@ -0,0 +1,295 @@
+/**
+ * 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 com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Preconditions;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ExecutorService;
+import org.apache.commons.lang3.RandomUtils;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.commons.lang3.tuple.Pair;
+import org.apache.pinot.common.metadata.controllerjob.ControllerJobType;
+import org.apache.pinot.common.metrics.ControllerMeter;
+import org.apache.pinot.common.metrics.ControllerMetrics;
+import org.apache.pinot.controller.ControllerConf;
+import org.apache.pinot.controller.LeadControllerManager;
+import org.apache.pinot.controller.helix.core.PinotHelixResourceManager;
+import org.apache.pinot.controller.helix.core.periodictask.ControllerPeriodicTask;
+import org.apache.pinot.spi.config.table.TableConfig;
+import org.apache.pinot.spi.utils.CommonConstants;
+import org.apache.pinot.spi.utils.JsonUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+/**
+ * Periodic task to check whether a user triggered rebalance job is completed or not, and retry if failed. The retry
+ * job is started with the same rebalance configs provided by the user and does best effort to stop the other jobs
+ * for the same table. This task can be configured to just check failures and report metrics, and not to do retry.
+ */
+public class RebalanceChecker extends ControllerPeriodicTask<Void> {
+ private static final Logger LOGGER = LoggerFactory.getLogger(RebalanceChecker.class);
+ private static final double RETRY_DELAY_SCALE_FACTOR = 2.0;
+ private final ExecutorService _executorService;
+
+ public RebalanceChecker(PinotHelixResourceManager pinotHelixResourceManager,
+ LeadControllerManager leadControllerManager, ControllerConf config, ControllerMetrics controllerMetrics,
+ ExecutorService executorService) {
+ super(RebalanceChecker.class.getSimpleName(), config.getRebalanceCheckerFrequencyInSeconds(),
+ config.getRebalanceCheckerInitialDelayInSeconds(), pinotHelixResourceManager, leadControllerManager,
+ controllerMetrics);
+ _executorService = executorService;
+ }
+
+ @Override
+ protected void processTable(String tableNameWithType) {
+ try {
+ LOGGER.info("Start to retry rebalance for table: {}", tableNameWithType);
+ // Get all rebalance jobs as tracked in ZK for this table to check if there is any failure to retry.
+ Map<String, Map<String, String>> allJobMetadata = _pinotHelixResourceManager.getAllJobsForTable(tableNameWithType,
Review Comment:
Property store is actually not cached, so let's make it one read for all tables
--
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
Re: [PR] Add support to retry failed table rebalance [pinot]
Posted by "klsince (via GitHub)" <gi...@apache.org>.
klsince commented on code in PR #11740:
URL: https://github.com/apache/pinot/pull/11740#discussion_r1361254945
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManager.java:
##########
@@ -2063,28 +2070,76 @@ public boolean addNewReloadAllSegmentsJob(String tableNameWithType, String jobId
ZKMetadataProvider.constructPropertyStorePathForControllerJob(ControllerJobType.RELOAD_SEGMENT));
}
+ /**
+ * Adds a new job metadata for controller job like table rebalance or reload into ZK
+ * @param jobId job's UUID
+ * @param jobMetadata the job metadata
+ * @param jobResourcePath zkPath to add the new job metadata
+ * @return boolean representing success / failure of the ZK write step
+ */
public boolean addControllerJobToZK(String jobId, Map<String, String> jobMetadata, String jobResourcePath) {
+ return addControllerJobToZK(jobId, jobMetadata, jobResourcePath, prev -> true);
+ }
+
+ /**
+ * Adds a new job metadata for controller job like table rebalance or reload into ZK
+ * @param jobId job's UUID
+ * @param jobMetadata the job metadata
+ * @param jobResourcePath zkPath to add the new job metadata
+ * @param prevJobMetadataChecker to check the previous job metadata before adding new one
+ * @return boolean representing success / failure of the ZK write step
+ */
+ public boolean addControllerJobToZK(String jobId, Map<String, String> jobMetadata, String jobResourcePath,
+ Predicate<Map<String, String>> prevJobMetadataChecker) {
Preconditions.checkState(jobMetadata.get(CommonConstants.ControllerJob.SUBMISSION_TIME_MS) != null,
"Submission Time in JobMetadata record not set. Cannot expire these records");
Stat stat = new Stat();
- ZNRecord tableJobsZnRecord = _propertyStore.get(jobResourcePath, stat, AccessOption.PERSISTENT);
- if (tableJobsZnRecord != null) {
- Map<String, Map<String, String>> tasks = tableJobsZnRecord.getMapFields();
- tasks.put(jobId, jobMetadata);
- if (tasks.size() > CommonConstants.ControllerJob.MAXIMUM_CONTROLLER_JOBS_IN_ZK) {
- tasks = tasks.entrySet().stream().sorted((v1, v2) -> Long.compare(
+ ZNRecord jobsZnRecord = _propertyStore.get(jobResourcePath, stat, AccessOption.PERSISTENT);
+ if (jobsZnRecord != null) {
+ Map<String, Map<String, String>> jobMetadataMap = jobsZnRecord.getMapFields();
+ Map<String, String> prevJobMetadata = jobMetadataMap.get(jobId);
+ if (!prevJobMetadataChecker.test(prevJobMetadata)) {
+ return false;
+ }
+ jobMetadataMap.put(jobId, jobMetadata);
+ if (jobMetadataMap.size() > CommonConstants.ControllerJob.MAXIMUM_CONTROLLER_JOBS_IN_ZK) {
+ jobMetadataMap = jobMetadataMap.entrySet().stream().sorted((v1, v2) -> Long.compare(
Long.parseLong(v2.getValue().get(CommonConstants.ControllerJob.SUBMISSION_TIME_MS)),
Long.parseLong(v1.getValue().get(CommonConstants.ControllerJob.SUBMISSION_TIME_MS))))
.collect(Collectors.toList()).subList(0, CommonConstants.ControllerJob.MAXIMUM_CONTROLLER_JOBS_IN_ZK)
.stream().collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
}
- tableJobsZnRecord.setMapFields(tasks);
- return _propertyStore.set(jobResourcePath, tableJobsZnRecord, stat.getVersion(), AccessOption.PERSISTENT);
+ jobsZnRecord.setMapFields(jobMetadataMap);
+ return _propertyStore.set(jobResourcePath, jobsZnRecord, stat.getVersion(), AccessOption.PERSISTENT);
} else {
- tableJobsZnRecord = new ZNRecord(jobResourcePath);
- tableJobsZnRecord.setMapField(jobId, jobMetadata);
- return _propertyStore.set(jobResourcePath, tableJobsZnRecord, AccessOption.PERSISTENT);
+ jobsZnRecord = new ZNRecord(jobResourcePath);
+ jobsZnRecord.setMapField(jobId, jobMetadata);
+ return _propertyStore.set(jobResourcePath, jobsZnRecord, AccessOption.PERSISTENT);
+ }
+ }
+
+ /**
+ * Update existing job metadata belong to the table
+ * @param tableNameWithType whose job metadata to be updated
+ * @param jobResourcePath zkPath to look for the job metadata for the table
+ * @param updater to modify the job metadata in place
+ * @return boolean representing success / failure of the ZK write step
+ */
+ public boolean updateAllJobsForTable(String tableNameWithType, String jobResourcePath,
Review Comment:
sgtm
--
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
Re: [PR] Add support to retry failed table rebalance [pinot]
Posted by "Jackie-Jiang (via GitHub)" <gi...@apache.org>.
Jackie-Jiang commented on code in PR #11740:
URL: https://github.com/apache/pinot/pull/11740#discussion_r1361177004
##########
pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotTableRestletResource.java:
##########
@@ -666,6 +678,92 @@ public RebalanceResult rebalance(
}
}
+ @DELETE
+ @Produces(MediaType.APPLICATION_JSON)
+ @Authenticate(AccessType.UPDATE)
+ @Path("/tables/{tableName}/rebalance")
+ @Authorize(targetType = TargetType.TABLE, paramName = "tableName", action = Actions.Table.ABORT_REBALANCE)
+ @ApiOperation(value = "Abort all rebalance jobs for the given table, and noop if no rebalance is running", notes =
+ "Abort all rebalance jobs for the given table, and noop if no rebalance is running")
+ public List<String> abortRebalance(
+ @ApiParam(value = "Name of the table to rebalance", required = true) @PathParam("tableName") String tableName,
+ @ApiParam(value = "OFFLINE|REALTIME", required = true) @QueryParam("type") String tableTypeStr) {
+ String tableNameWithType = constructTableNameWithType(tableName, tableTypeStr);
+ String zkPath = ZKMetadataProvider.constructPropertyStorePathForControllerJob(ControllerJobType.TABLE_REBALANCE);
+ List<String> abortedJobIds = new ArrayList<>();
+ // The rebalance job updates progress status kept in ZK regularly. If it sees ABORTED status, it aborts itself.
+ boolean updated = _pinotHelixResourceManager.updateAllJobsForTable(tableNameWithType, zkPath, jobMetadata -> {
+ String jobId = jobMetadata.get(CommonConstants.ControllerJob.JOB_ID);
+ try {
+ String jobStatsInStr = jobMetadata.get(RebalanceJobConstants.JOB_METADATA_KEY_REBALANCE_PROGRESS_STATS);
+ TableRebalanceProgressStats jobStats =
+ JsonUtils.stringToObject(jobStatsInStr, TableRebalanceProgressStats.class);
+ if (jobStats.getStatus() != RebalanceResult.Status.IN_PROGRESS) {
+ return;
+ }
+ abortedJobIds.add(jobId);
+ LOGGER.info("Abort rebalance job: {} for table: {}", jobId, tableNameWithType);
+ jobStats.setStatus(RebalanceResult.Status.ABORTED);
+ jobMetadata.put(RebalanceJobConstants.JOB_METADATA_KEY_REBALANCE_PROGRESS_STATS,
+ JsonUtils.objectToString(jobStats));
+
+ // Also cancel following retries for the rebalance job.
+ String jobCtxInStr = jobMetadata.get(RebalanceJobConstants.JOB_METADATA_KEY_REBALANCE_ATTEMPT_CONTEXT);
+ if (StringUtils.isEmpty(jobCtxInStr)) {
+ // In case the job is submitted by older code, it may not have job ctx field.
+ return;
+ }
+ TableRebalanceAttemptContext jobCtx = JsonUtils.stringToObject(jobCtxInStr, TableRebalanceAttemptContext.class);
+ jobCtx.setCancelRetry(true);
Review Comment:
Do we need this extra flag? IMO we should use the rebalance status to mark whether we need retry. If `ABORTED` doesn't always mean no retry, we can introduce `CANCELLED`
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManager.java:
##########
@@ -2063,28 +2070,76 @@ public boolean addNewReloadAllSegmentsJob(String tableNameWithType, String jobId
ZKMetadataProvider.constructPropertyStorePathForControllerJob(ControllerJobType.RELOAD_SEGMENT));
}
+ /**
+ * Adds a new job metadata for controller job like table rebalance or reload into ZK
+ * @param jobId job's UUID
+ * @param jobMetadata the job metadata
+ * @param jobResourcePath zkPath to add the new job metadata
+ * @return boolean representing success / failure of the ZK write step
+ */
public boolean addControllerJobToZK(String jobId, Map<String, String> jobMetadata, String jobResourcePath) {
Review Comment:
Not introduced in this PR, but suggest changing the last argument to `ControllerJobType`
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/rebalance/RebalanceJobConstants.java:
##########
@@ -24,4 +24,6 @@ private RebalanceJobConstants() {
// Progress status of the rebalance operartion
public static final String JOB_METADATA_KEY_REBALANCE_PROGRESS_STATS = "REBALANCE_PROGRESS_STATS";
+ // Configs to retry the rebalance operartion
+ public static final String JOB_METADATA_KEY_REBALANCE_ATTEMPT_CONTEXT = "REBALANCE_ATTEMPT_CONTEXT";
Review Comment:
Just call it `REBALANCE_CONTEXT`?
##########
pinot-core/src/main/java/org/apache/pinot/core/auth/Actions.java:
##########
@@ -89,6 +89,7 @@ public static class Cluster {
// Action names for table
public static class Table {
+ public static final String ABORT_REBALANCE = "AbortRebalance";
Review Comment:
We might want to call it `CancelRebalance` (we have `CancelQuery`)
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManager.java:
##########
@@ -1992,19 +1994,24 @@ public Map<String, Map<String, String>> getAllJobsForTable(String tableNameWithT
for (ControllerJobType jobType : jobTypes) {
String jobsResourcePath = ZKMetadataProvider.constructPropertyStorePathForControllerJob(jobType);
try {
+ if (!_propertyStore.exists(jobsResourcePath, -1)) {
Review Comment:
This check is actually redundant. The null check itself is good enough
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManager.java:
##########
@@ -2063,28 +2070,76 @@ public boolean addNewReloadAllSegmentsJob(String tableNameWithType, String jobId
ZKMetadataProvider.constructPropertyStorePathForControllerJob(ControllerJobType.RELOAD_SEGMENT));
}
+ /**
+ * Adds a new job metadata for controller job like table rebalance or reload into ZK
+ * @param jobId job's UUID
+ * @param jobMetadata the job metadata
+ * @param jobResourcePath zkPath to add the new job metadata
+ * @return boolean representing success / failure of the ZK write step
+ */
public boolean addControllerJobToZK(String jobId, Map<String, String> jobMetadata, String jobResourcePath) {
+ return addControllerJobToZK(jobId, jobMetadata, jobResourcePath, prev -> true);
+ }
+
+ /**
+ * Adds a new job metadata for controller job like table rebalance or reload into ZK
+ * @param jobId job's UUID
+ * @param jobMetadata the job metadata
+ * @param jobResourcePath zkPath to add the new job metadata
+ * @param prevJobMetadataChecker to check the previous job metadata before adding new one
+ * @return boolean representing success / failure of the ZK write step
+ */
+ public boolean addControllerJobToZK(String jobId, Map<String, String> jobMetadata, String jobResourcePath,
Review Comment:
Suggest changing `jobResourcePath` to `ControllerJobType`
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManager.java:
##########
@@ -1992,19 +1994,24 @@ public Map<String, Map<String, String>> getAllJobsForTable(String tableNameWithT
for (ControllerJobType jobType : jobTypes) {
String jobsResourcePath = ZKMetadataProvider.constructPropertyStorePathForControllerJob(jobType);
try {
+ if (!_propertyStore.exists(jobsResourcePath, -1)) {
Review Comment:
(minor) Use `AccessOption.PERSISTENT`, same for `get()`
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/rebalance/RebalanceChecker.java:
##########
@@ -0,0 +1,330 @@
+/**
+ * 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 com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Preconditions;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.Set;
+import java.util.concurrent.ExecutorService;
+import org.apache.commons.lang3.RandomUtils;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.commons.lang3.tuple.Pair;
+import org.apache.pinot.common.metadata.ZKMetadataProvider;
+import org.apache.pinot.common.metadata.controllerjob.ControllerJobType;
+import org.apache.pinot.common.metrics.ControllerGauge;
+import org.apache.pinot.common.metrics.ControllerMeter;
+import org.apache.pinot.common.metrics.ControllerMetrics;
+import org.apache.pinot.controller.ControllerConf;
+import org.apache.pinot.controller.LeadControllerManager;
+import org.apache.pinot.controller.helix.core.PinotHelixResourceManager;
+import org.apache.pinot.controller.helix.core.periodictask.ControllerPeriodicTask;
+import org.apache.pinot.spi.config.table.TableConfig;
+import org.apache.pinot.spi.utils.CommonConstants;
+import org.apache.pinot.spi.utils.JsonUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+/**
+ * Periodic task to check whether a user triggered rebalance job is completed or not, and retry if failed. The retry
+ * job is started with the same rebalance configs provided by the user and does best effort to stop the other jobs
+ * for the same table. This task can be configured to just check failures and report metrics, and not to do retry.
+ */
+public class RebalanceChecker extends ControllerPeriodicTask<Void> {
+ private static final Logger LOGGER = LoggerFactory.getLogger(RebalanceChecker.class);
+ private static final double RETRY_DELAY_SCALE_FACTOR = 2.0;
+ private final ExecutorService _executorService;
+
+ public RebalanceChecker(PinotHelixResourceManager pinotHelixResourceManager,
+ LeadControllerManager leadControllerManager, ControllerConf config, ControllerMetrics controllerMetrics,
+ ExecutorService executorService) {
+ super(RebalanceChecker.class.getSimpleName(), config.getRebalanceCheckerFrequencyInSeconds(),
+ config.getRebalanceCheckerInitialDelayInSeconds(), pinotHelixResourceManager, leadControllerManager,
+ controllerMetrics);
+ _executorService = executorService;
+ }
+
+ protected void processTables(List<String> tableNamesWithType, Properties periodicTaskProperties) {
+ int numTables = tableNamesWithType.size();
+ LOGGER.info("Processing {} tables in task: {}", numTables, _taskName);
+ int numTablesProcessed = 0;
+ for (String tableNameWithType : tableNamesWithType) {
+ if (!isStarted()) {
+ LOGGER.info("Task: {} is stopped, early terminate the task", _taskName);
+ break;
+ }
+ try {
+ processTable(tableNameWithType);
Review Comment:
We are still doing per-table check here, which means if there are 1000 tables, we will read table config and job metadata 1000 times. Instead, we should read job metadata only once, and gather the tables needed to be retried, then schedule the retries
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/rebalance/TableRebalancer.java:
##########
@@ -447,6 +467,9 @@ public RebalanceResult rebalance(TableConfig tableConfig, RebalanceConfig rebala
// Check version and update IdealState
try {
+ if (_tableRebalanceObserver.isAborted()) {
+ throw new RuntimeException(String.format("Rebalance for table: %s has been aborted", tableNameWithType));
Review Comment:
Will this cause `RebalanceResult` to be `UNKNOWN_ERROR`? We should make it explicit that it is cancelled
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManager.java:
##########
@@ -2063,28 +2070,76 @@ public boolean addNewReloadAllSegmentsJob(String tableNameWithType, String jobId
ZKMetadataProvider.constructPropertyStorePathForControllerJob(ControllerJobType.RELOAD_SEGMENT));
}
+ /**
+ * Adds a new job metadata for controller job like table rebalance or reload into ZK
+ * @param jobId job's UUID
+ * @param jobMetadata the job metadata
+ * @param jobResourcePath zkPath to add the new job metadata
+ * @return boolean representing success / failure of the ZK write step
+ */
public boolean addControllerJobToZK(String jobId, Map<String, String> jobMetadata, String jobResourcePath) {
+ return addControllerJobToZK(jobId, jobMetadata, jobResourcePath, prev -> true);
+ }
+
+ /**
+ * Adds a new job metadata for controller job like table rebalance or reload into ZK
+ * @param jobId job's UUID
+ * @param jobMetadata the job metadata
+ * @param jobResourcePath zkPath to add the new job metadata
+ * @param prevJobMetadataChecker to check the previous job metadata before adding new one
+ * @return boolean representing success / failure of the ZK write step
+ */
+ public boolean addControllerJobToZK(String jobId, Map<String, String> jobMetadata, String jobResourcePath,
+ Predicate<Map<String, String>> prevJobMetadataChecker) {
Preconditions.checkState(jobMetadata.get(CommonConstants.ControllerJob.SUBMISSION_TIME_MS) != null,
"Submission Time in JobMetadata record not set. Cannot expire these records");
Stat stat = new Stat();
- ZNRecord tableJobsZnRecord = _propertyStore.get(jobResourcePath, stat, AccessOption.PERSISTENT);
- if (tableJobsZnRecord != null) {
- Map<String, Map<String, String>> tasks = tableJobsZnRecord.getMapFields();
- tasks.put(jobId, jobMetadata);
- if (tasks.size() > CommonConstants.ControllerJob.MAXIMUM_CONTROLLER_JOBS_IN_ZK) {
- tasks = tasks.entrySet().stream().sorted((v1, v2) -> Long.compare(
+ ZNRecord jobsZnRecord = _propertyStore.get(jobResourcePath, stat, AccessOption.PERSISTENT);
+ if (jobsZnRecord != null) {
+ Map<String, Map<String, String>> jobMetadataMap = jobsZnRecord.getMapFields();
+ Map<String, String> prevJobMetadata = jobMetadataMap.get(jobId);
+ if (!prevJobMetadataChecker.test(prevJobMetadata)) {
+ return false;
+ }
+ jobMetadataMap.put(jobId, jobMetadata);
+ if (jobMetadataMap.size() > CommonConstants.ControllerJob.MAXIMUM_CONTROLLER_JOBS_IN_ZK) {
+ jobMetadataMap = jobMetadataMap.entrySet().stream().sorted((v1, v2) -> Long.compare(
Long.parseLong(v2.getValue().get(CommonConstants.ControllerJob.SUBMISSION_TIME_MS)),
Long.parseLong(v1.getValue().get(CommonConstants.ControllerJob.SUBMISSION_TIME_MS))))
.collect(Collectors.toList()).subList(0, CommonConstants.ControllerJob.MAXIMUM_CONTROLLER_JOBS_IN_ZK)
.stream().collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
}
- tableJobsZnRecord.setMapFields(tasks);
- return _propertyStore.set(jobResourcePath, tableJobsZnRecord, stat.getVersion(), AccessOption.PERSISTENT);
+ jobsZnRecord.setMapFields(jobMetadataMap);
+ return _propertyStore.set(jobResourcePath, jobsZnRecord, stat.getVersion(), AccessOption.PERSISTENT);
} else {
- tableJobsZnRecord = new ZNRecord(jobResourcePath);
- tableJobsZnRecord.setMapField(jobId, jobMetadata);
- return _propertyStore.set(jobResourcePath, tableJobsZnRecord, AccessOption.PERSISTENT);
+ jobsZnRecord = new ZNRecord(jobResourcePath);
+ jobsZnRecord.setMapField(jobId, jobMetadata);
+ return _propertyStore.set(jobResourcePath, jobsZnRecord, AccessOption.PERSISTENT);
+ }
+ }
+
+ /**
+ * Update existing job metadata belong to the table
+ * @param tableNameWithType whose job metadata to be updated
+ * @param jobResourcePath zkPath to look for the job metadata for the table
+ * @param updater to modify the job metadata in place
+ * @return boolean representing success / failure of the ZK write step
+ */
+ public boolean updateAllJobsForTable(String tableNameWithType, String jobResourcePath,
Review Comment:
Change `jobResourcePath` to `ControllerJobType`, also suggest making the updater on the whole `jobMetadataMap` which is more general. Currently it relies on the job id to be encoded in the value
(minor) Suggest renaming the method to `updateJobsForTable` since it only touches one job type
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/rebalance/TableRebalanceAttemptContext.java:
##########
@@ -0,0 +1,108 @@
+/**
+ * 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 com.google.common.annotations.VisibleForTesting;
+
+
+/**
+ * Track the job configs and attempt numbers as part of the job ZK metadata for future retry.
+ */
+public class TableRebalanceAttemptContext {
Review Comment:
Suggest just naming it `TableRebalanceContext`. Retry is just one usage of this context, and we may add more usage in the future (e.g. `jobId` is not only used in retry)
--
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
Re: [PR] Add support to retry failed table rebalance [pinot]
Posted by "klsince (via GitHub)" <gi...@apache.org>.
klsince merged PR #11740:
URL: https://github.com/apache/pinot/pull/11740
--
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
Re: [PR] Add support to retry failed table rebalance [pinot]
Posted by "klsince (via GitHub)" <gi...@apache.org>.
klsince commented on code in PR #11740:
URL: https://github.com/apache/pinot/pull/11740#discussion_r1355861888
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/rebalance/TableRebalancer.java:
##########
@@ -137,6 +139,23 @@ public static String createUniqueRebalanceJobIdentifier() {
}
public RebalanceResult rebalance(TableConfig tableConfig, RebalanceConfig rebalanceConfig) {
+ long startTime = System.currentTimeMillis();
+ String tableNameWithType = tableConfig.getTableName();
+ String status = "ABORTED";
Review Comment:
right, this was for uncaught exceptions. I thought about `unknown`, but `errored` seems better.
--
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
Re: [PR] Add support to retry failed table rebalance [pinot]
Posted by "klsince (via GitHub)" <gi...@apache.org>.
klsince commented on code in PR #11740:
URL: https://github.com/apache/pinot/pull/11740#discussion_r1361217787
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/rebalance/RebalanceChecker.java:
##########
@@ -0,0 +1,330 @@
+/**
+ * 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 com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Preconditions;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.Set;
+import java.util.concurrent.ExecutorService;
+import org.apache.commons.lang3.RandomUtils;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.commons.lang3.tuple.Pair;
+import org.apache.pinot.common.metadata.ZKMetadataProvider;
+import org.apache.pinot.common.metadata.controllerjob.ControllerJobType;
+import org.apache.pinot.common.metrics.ControllerGauge;
+import org.apache.pinot.common.metrics.ControllerMeter;
+import org.apache.pinot.common.metrics.ControllerMetrics;
+import org.apache.pinot.controller.ControllerConf;
+import org.apache.pinot.controller.LeadControllerManager;
+import org.apache.pinot.controller.helix.core.PinotHelixResourceManager;
+import org.apache.pinot.controller.helix.core.periodictask.ControllerPeriodicTask;
+import org.apache.pinot.spi.config.table.TableConfig;
+import org.apache.pinot.spi.utils.CommonConstants;
+import org.apache.pinot.spi.utils.JsonUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+/**
+ * Periodic task to check whether a user triggered rebalance job is completed or not, and retry if failed. The retry
+ * job is started with the same rebalance configs provided by the user and does best effort to stop the other jobs
+ * for the same table. This task can be configured to just check failures and report metrics, and not to do retry.
+ */
+public class RebalanceChecker extends ControllerPeriodicTask<Void> {
+ private static final Logger LOGGER = LoggerFactory.getLogger(RebalanceChecker.class);
+ private static final double RETRY_DELAY_SCALE_FACTOR = 2.0;
+ private final ExecutorService _executorService;
+
+ public RebalanceChecker(PinotHelixResourceManager pinotHelixResourceManager,
+ LeadControllerManager leadControllerManager, ControllerConf config, ControllerMetrics controllerMetrics,
+ ExecutorService executorService) {
+ super(RebalanceChecker.class.getSimpleName(), config.getRebalanceCheckerFrequencyInSeconds(),
+ config.getRebalanceCheckerInitialDelayInSeconds(), pinotHelixResourceManager, leadControllerManager,
+ controllerMetrics);
+ _executorService = executorService;
+ }
+
+ protected void processTables(List<String> tableNamesWithType, Properties periodicTaskProperties) {
+ int numTables = tableNamesWithType.size();
+ LOGGER.info("Processing {} tables in task: {}", numTables, _taskName);
+ int numTablesProcessed = 0;
+ for (String tableNameWithType : tableNamesWithType) {
+ if (!isStarted()) {
+ LOGGER.info("Task: {} is stopped, early terminate the task", _taskName);
+ break;
+ }
+ try {
+ processTable(tableNameWithType);
Review Comment:
> ... job metadata 1000 times
there is only one ZNode to keep all rebalance job metadata from all tables and there is cache in zkClient, so it's not 1000 reads against ZK. But I should delay access to tableConfig after the table is decided for another retry.
--
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
Re: [PR] Add support to retry failed table rebalance [pinot]
Posted by "klsince (via GitHub)" <gi...@apache.org>.
klsince commented on code in PR #11740:
URL: https://github.com/apache/pinot/pull/11740#discussion_r1361219867
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/rebalance/TableRebalancer.java:
##########
@@ -447,6 +467,9 @@ public RebalanceResult rebalance(TableConfig tableConfig, RebalanceConfig rebala
// Check version and update IdealState
try {
+ if (_tableRebalanceObserver.isAborted()) {
+ throw new RuntimeException(String.format("Rebalance for table: %s has been aborted", tableNameWithType));
Review Comment:
this leads to FAILED status, as I put the abort checks inside try-catch blocks in TableRebalancer. the UNKNOWN_ERROR only happens when there is unexpected (uncaught) exceptions, and such case is tracked by an error counter to set alert on if needed.
--
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
Re: [PR] Add support to retry failed table rebalance [pinot]
Posted by "klsince (via GitHub)" <gi...@apache.org>.
klsince commented on PR #11740:
URL: https://github.com/apache/pinot/pull/11740#issuecomment-1747888540
> One concern I have is: if the initial rebalance does not succeed because too many helix messages flood servers, retry will improve workload on servers, which may lead to more severe problem. To alleviate this issue,
>
> 1. is there a way to stop the failed or running table rebalance?
> 2. is there a way to drop existing table rebalance helix messages?
Good questions!
for 1: I added a sanity check when rebalance job updates its job progress status in ZK. If the job progress is FAILED, it aborts self. With that, before kicking off a retry, the RebalanceChecker aborts existing jobs by setting their status to FAILED.
for 2. I’m not aware of way to clean up helix msg. Because retry continues the rebablance from where the ideal state is, there wouldn’t be too many redundant helix transition msgs iiuc.
But your concern reminds me that I should add retry backoff and jitters so that RebalanceChecker doesn’t retry too soon after last failure and avoid retrying many tables all at once, flooding the servers.
--
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
Re: [PR] Add support to retry failed table rebalance [pinot]
Posted by "klsince (via GitHub)" <gi...@apache.org>.
klsince commented on code in PR #11740:
URL: https://github.com/apache/pinot/pull/11740#discussion_r1355848314
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/rebalance/ZkBasedTableRebalanceObserver.java:
##########
@@ -98,10 +117,9 @@ private void updateOnStart(Map<String, Map<String, String>> currentState,
@Override
public void onSuccess(String msg) {
- Preconditions.checkState(_tableRebalanceProgressStats.getStatus() != RebalanceResult.Status.DONE.toString(),
+ Preconditions.checkState(!RebalanceResult.Status.DONE.toString().equals(_tableRebalanceProgressStats.getStatus()),
Review Comment:
sgtm
--
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
Re: [PR] Add support to retry failed table rebalance [pinot]
Posted by "klsince (via GitHub)" <gi...@apache.org>.
klsince commented on code in PR #11740:
URL: https://github.com/apache/pinot/pull/11740#discussion_r1355840502
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/rebalance/RebalanceChecker.java:
##########
@@ -0,0 +1,279 @@
+/**
+ * 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 com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Preconditions;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ExecutorService;
+import org.apache.commons.lang3.RandomUtils;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.commons.lang3.tuple.Pair;
+import org.apache.pinot.common.metadata.ZKMetadataProvider;
+import org.apache.pinot.common.metadata.controllerjob.ControllerJobType;
+import org.apache.pinot.common.metrics.ControllerMeter;
+import org.apache.pinot.common.metrics.ControllerMetrics;
+import org.apache.pinot.controller.ControllerConf;
+import org.apache.pinot.controller.LeadControllerManager;
+import org.apache.pinot.controller.helix.core.PinotHelixResourceManager;
+import org.apache.pinot.controller.helix.core.periodictask.ControllerPeriodicTask;
+import org.apache.pinot.spi.config.table.TableConfig;
+import org.apache.pinot.spi.utils.CommonConstants;
+import org.apache.pinot.spi.utils.JsonUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+/**
+ * Periodic task to check whether a user triggered rebalance job is completed or not, and retry if failed. The retry
+ * job is started with the same rebalance configs provided by the user and does best effort to stop the other jobs
+ * for the same table. This task can be configured to just check failures and report metrics, and not to do retry.
+ */
+public class RebalanceChecker extends ControllerPeriodicTask<Void> {
+ private static final Logger LOGGER = LoggerFactory.getLogger(RebalanceChecker.class);
+ private static final double RETRY_DELAY_SCALE_FACTOR = 2.0;
+ private final boolean _checkOnly;
+ private final ExecutorService _executorService;
+
+ public RebalanceChecker(PinotHelixResourceManager pinotHelixResourceManager,
+ LeadControllerManager leadControllerManager, ControllerConf config, ControllerMetrics controllerMetrics,
+ ExecutorService executorService) {
+ super(RebalanceChecker.class.getSimpleName(), config.getRebalanceCheckerFrequencyInSeconds(),
+ config.getRebalanceCheckerInitialDelayInSeconds(), pinotHelixResourceManager, leadControllerManager,
+ controllerMetrics);
+ _checkOnly = config.isRebalanceCheckerCheckOnly();
+ _executorService = executorService;
+ }
+
+ @Override
+ protected void processTable(String tableNameWithType) {
+ _executorService.submit(() -> {
Review Comment:
yup, good point!
--
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
Re: [PR] Add support to retry failed table rebalance [pinot]
Posted by "klsince (via GitHub)" <gi...@apache.org>.
klsince commented on code in PR #11740:
URL: https://github.com/apache/pinot/pull/11740#discussion_r1355895061
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/rebalance/TableRebalanceRetryConfig.java:
##########
@@ -0,0 +1,74 @@
+/**
+ * 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;
+
+/**
+ * Track the job configs and retry numbers as part of the job ZK metadata for future retry.
+ */
+public class TableRebalanceRetryConfig {
Review Comment:
https://github.com/apache/pinot/pull/11790 to refactor jobId out of RebalanceConfig. pls review
--
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
Re: [PR] Add support to retry failed table rebalance [pinot]
Posted by "klsince (via GitHub)" <gi...@apache.org>.
klsince commented on code in PR #11740:
URL: https://github.com/apache/pinot/pull/11740#discussion_r1355843815
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/rebalance/TableRebalanceRetryConfig.java:
##########
@@ -0,0 +1,74 @@
+/**
+ * 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;
+
+/**
+ * Track the job configs and retry numbers as part of the job ZK metadata for future retry.
+ */
+public class TableRebalanceRetryConfig {
Review Comment:
sgtm, renamed to TableRebalanceContext.
But moving `jobId` out of `RebalanceConfig` would require changes on PinotHelixResourceManager.rebalanceTable() and TableRebalancer.rebalance() to accept `jobId` along with `RebalanceConfig`. If you feel strong about this, I can open a PR to refactor on this firstly.
--
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
Re: [PR] Add support to retry failed table rebalance [pinot]
Posted by "klsince (via GitHub)" <gi...@apache.org>.
klsince commented on code in PR #11740:
URL: https://github.com/apache/pinot/pull/11740#discussion_r1361245932
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManager.java:
##########
@@ -1992,19 +1994,24 @@ public Map<String, Map<String, String>> getAllJobsForTable(String tableNameWithT
for (ControllerJobType jobType : jobTypes) {
String jobsResourcePath = ZKMetadataProvider.constructPropertyStorePathForControllerJob(jobType);
try {
+ if (!_propertyStore.exists(jobsResourcePath, -1)) {
Review Comment:
as tested, propertyStore.get() threw ZkNoNodeException. The current try-catch would continue the for-loop upon this exception, but the warn logs were kinda annoying, so I added the explicit existence check.
looks like no AccessOption is `-1`, but changed to PERSISTENT anyway
Updated: as it turned out, using `-1` as the option to get ZNode was why it threw exception, because `-1` would set this specific option to true: `public static int THROW_EXCEPTION_IFNOTEXIST = 16;`.
--
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
Re: [PR] Add support to retry failed table rebalance [pinot]
Posted by "Jackie-Jiang (via GitHub)" <gi...@apache.org>.
Jackie-Jiang commented on code in PR #11740:
URL: https://github.com/apache/pinot/pull/11740#discussion_r1361297839
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/rebalance/TableRebalancer.java:
##########
@@ -447,6 +467,9 @@ public RebalanceResult rebalance(TableConfig tableConfig, RebalanceConfig rebala
// Check version and update IdealState
try {
+ if (_tableRebalanceObserver.isAborted()) {
+ throw new RuntimeException(String.format("Rebalance for table: %s has been aborted", tableNameWithType));
Review Comment:
This is regular flow. Can we make it returning CANCELLED explicitly?
--
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
Re: [PR] Add support to retry failed table rebalance [pinot]
Posted by "zhtaoxiang (via GitHub)" <gi...@apache.org>.
zhtaoxiang commented on code in PR #11740:
URL: https://github.com/apache/pinot/pull/11740#discussion_r1348079942
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManager.java:
##########
@@ -2063,28 +2066,76 @@ public boolean addNewReloadAllSegmentsJob(String tableNameWithType, String jobId
ZKMetadataProvider.constructPropertyStorePathForControllerJob(ControllerJobType.RELOAD_SEGMENT));
}
+ /**
+ * Adds a new job metadata for controller job like table rebalance or reload into ZK
+ * @param jobId job's UUID
+ * @param jobMetadata the job metadata
+ * @param jobResourcePath zkPath to add the new job metadata
+ * @return boolean representing success / failure of the ZK write step
+ */
public boolean addControllerJobToZK(String jobId, Map<String, String> jobMetadata, String jobResourcePath) {
+ return addControllerJobToZK(jobId, jobMetadata, jobResourcePath, prev -> true);
+ }
+
+ /**
+ * Adds a new job metadata for controller job like table rebalance or reload into ZK
+ * @param jobId job's UUID
+ * @param jobMetadata the job metadata
+ * @param jobResourcePath zkPath to add the new job metadata
+ * @param prevJobMetadataChecker to check the previous job metadata before adding new one
Review Comment:
could we add some explanation on what specifically to check?
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/rebalance/ZkBasedTableRebalanceObserver.java:
##########
@@ -126,30 +143,74 @@ public int getNumUpdatesToZk() {
return _numUpdatesToZk;
}
+ @VisibleForTesting
+ TableRebalanceRetryConfig getTableRebalanceJobRetryConfig() {
+ return _tableRebalanceJobRetryConfig;
+ }
+
private void trackStatsInZk() {
+ Map<String, String> jobMetadata =
+ createJobMetadata(_tableNameWithType, _rebalanceJobId, _tableRebalanceProgressStats,
+ _tableRebalanceJobRetryConfig);
+ _pinotHelixResourceManager.addControllerJobToZK(_rebalanceJobId, jobMetadata,
+ ZKMetadataProvider.constructPropertyStorePathForControllerJob(ControllerJobType.TABLE_REBALANCE),
+ prevJobMetadata -> {
+ // Abort the job when we're sure it has failed, otherwise continue to update the status.
+ if (prevJobMetadata == null) {
+ return true;
+ }
+ String prevStatsInStr = prevJobMetadata.get(RebalanceJobConstants.JOB_METADATA_KEY_REBALANCE_PROGRESS_STATS);
+ TableRebalanceProgressStats prevStats;
+ try {
+ prevStats = JsonUtils.stringToObject(prevStatsInStr, TableRebalanceProgressStats.class);
+ } catch (JsonProcessingException e) {
+ throw new RuntimeException(e);
+ }
Review Comment:
should we log the exception and return false?
##########
docker/images/pinot/etc/jmx_prometheus_javaagent/configs/controller.yml:
##########
@@ -100,6 +100,12 @@ rules:
labels:
table: "$1"
taskType: "$2"
+- pattern: "\"org.apache.pinot.common.metrics\"<type=\"ControllerMetrics\", name=\"pinot.controller.([^\\.]*?)\\.(\\w+).tableRebalanceExecutionTimeMs\"><>(\\w+)"
Review Comment:
we also need to add this to the pinot.yml file
--
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
Re: [PR] Add support to retry failed table rebalance [pinot]
Posted by "klsince (via GitHub)" <gi...@apache.org>.
klsince commented on code in PR #11740:
URL: https://github.com/apache/pinot/pull/11740#discussion_r1348093609
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/rebalance/ZkBasedTableRebalanceObserver.java:
##########
@@ -126,30 +143,74 @@ public int getNumUpdatesToZk() {
return _numUpdatesToZk;
}
+ @VisibleForTesting
+ TableRebalanceRetryConfig getTableRebalanceJobRetryConfig() {
+ return _tableRebalanceJobRetryConfig;
+ }
+
private void trackStatsInZk() {
+ Map<String, String> jobMetadata =
+ createJobMetadata(_tableNameWithType, _rebalanceJobId, _tableRebalanceProgressStats,
+ _tableRebalanceJobRetryConfig);
+ _pinotHelixResourceManager.addControllerJobToZK(_rebalanceJobId, jobMetadata,
+ ZKMetadataProvider.constructPropertyStorePathForControllerJob(ControllerJobType.TABLE_REBALANCE),
+ prevJobMetadata -> {
+ // Abort the job when we're sure it has failed, otherwise continue to update the status.
+ if (prevJobMetadata == null) {
+ return true;
+ }
+ String prevStatsInStr = prevJobMetadata.get(RebalanceJobConstants.JOB_METADATA_KEY_REBALANCE_PROGRESS_STATS);
+ TableRebalanceProgressStats prevStats;
+ try {
+ prevStats = JsonUtils.stringToObject(prevStatsInStr, TableRebalanceProgressStats.class);
+ } catch (JsonProcessingException e) {
+ throw new RuntimeException(e);
+ }
Review Comment:
oh, good catch, this should be `return true`, to let current rebalance continue. As commented above, we abort if we really get a FAILED status, otherwise, just be conservative and let it continue.
--
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
Re: [PR] Add support to retry failed table rebalance [pinot]
Posted by "klsince (via GitHub)" <gi...@apache.org>.
klsince commented on code in PR #11740:
URL: https://github.com/apache/pinot/pull/11740#discussion_r1361214644
##########
pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotTableRestletResource.java:
##########
@@ -666,6 +678,92 @@ public RebalanceResult rebalance(
}
}
+ @DELETE
+ @Produces(MediaType.APPLICATION_JSON)
+ @Authenticate(AccessType.UPDATE)
+ @Path("/tables/{tableName}/rebalance")
+ @Authorize(targetType = TargetType.TABLE, paramName = "tableName", action = Actions.Table.ABORT_REBALANCE)
+ @ApiOperation(value = "Abort all rebalance jobs for the given table, and noop if no rebalance is running", notes =
+ "Abort all rebalance jobs for the given table, and noop if no rebalance is running")
+ public List<String> abortRebalance(
+ @ApiParam(value = "Name of the table to rebalance", required = true) @PathParam("tableName") String tableName,
+ @ApiParam(value = "OFFLINE|REALTIME", required = true) @QueryParam("type") String tableTypeStr) {
+ String tableNameWithType = constructTableNameWithType(tableName, tableTypeStr);
+ String zkPath = ZKMetadataProvider.constructPropertyStorePathForControllerJob(ControllerJobType.TABLE_REBALANCE);
+ List<String> abortedJobIds = new ArrayList<>();
+ // The rebalance job updates progress status kept in ZK regularly. If it sees ABORTED status, it aborts itself.
+ boolean updated = _pinotHelixResourceManager.updateAllJobsForTable(tableNameWithType, zkPath, jobMetadata -> {
+ String jobId = jobMetadata.get(CommonConstants.ControllerJob.JOB_ID);
+ try {
+ String jobStatsInStr = jobMetadata.get(RebalanceJobConstants.JOB_METADATA_KEY_REBALANCE_PROGRESS_STATS);
+ TableRebalanceProgressStats jobStats =
+ JsonUtils.stringToObject(jobStatsInStr, TableRebalanceProgressStats.class);
+ if (jobStats.getStatus() != RebalanceResult.Status.IN_PROGRESS) {
+ return;
+ }
+ abortedJobIds.add(jobId);
+ LOGGER.info("Abort rebalance job: {} for table: {}", jobId, tableNameWithType);
+ jobStats.setStatus(RebalanceResult.Status.ABORTED);
+ jobMetadata.put(RebalanceJobConstants.JOB_METADATA_KEY_REBALANCE_PROGRESS_STATS,
+ JsonUtils.objectToString(jobStats));
+
+ // Also cancel following retries for the rebalance job.
+ String jobCtxInStr = jobMetadata.get(RebalanceJobConstants.JOB_METADATA_KEY_REBALANCE_ATTEMPT_CONTEXT);
+ if (StringUtils.isEmpty(jobCtxInStr)) {
+ // In case the job is submitted by older code, it may not have job ctx field.
+ return;
+ }
+ TableRebalanceAttemptContext jobCtx = JsonUtils.stringToObject(jobCtxInStr, TableRebalanceAttemptContext.class);
+ jobCtx.setCancelRetry(true);
Review Comment:
ABORTED is used in two places: 1) here, we want to stop all future retries as well; 2) when next retry aborts previous ones, where we want retry continues until maxAttempts. I'm +1 to add CANCELLED for clarity.
--
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
Re: [PR] Add support to retry failed table rebalance [pinot]
Posted by "klsince (via GitHub)" <gi...@apache.org>.
klsince commented on code in PR #11740:
URL: https://github.com/apache/pinot/pull/11740#discussion_r1355848314
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/rebalance/ZkBasedTableRebalanceObserver.java:
##########
@@ -98,10 +117,9 @@ private void updateOnStart(Map<String, Map<String, String>> currentState,
@Override
public void onSuccess(String msg) {
- Preconditions.checkState(_tableRebalanceProgressStats.getStatus() != RebalanceResult.Status.DONE.toString(),
+ Preconditions.checkState(!RebalanceResult.Status.DONE.toString().equals(_tableRebalanceProgressStats.getStatus()),
Review Comment:
sgtm, will see if der/se works as expected with enum
--
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
Re: [PR] Add support to retry failed table rebalance [pinot]
Posted by "klsince (via GitHub)" <gi...@apache.org>.
klsince commented on code in PR #11740:
URL: https://github.com/apache/pinot/pull/11740#discussion_r1347843487
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/rebalance/TableRebalancer.java:
##########
@@ -137,6 +140,23 @@ public static String createUniqueRebalanceJobIdentifier() {
}
public RebalanceResult rebalance(TableConfig tableConfig, RebalanceConfig rebalanceConfig) {
+ long startTime = System.currentTimeMillis();
+ String tableNameWithType = tableConfig.getTableName();
+ try {
+ RebalanceResult result = doRebalance(tableConfig, rebalanceConfig);
+ if (_controllerMetrics != null && result.getStatus() == RebalanceResult.Status.FAILED) {
Review Comment:
sgtm, added result status as part of metric name.
--
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
Re: [PR] Add support to retry failed table rebalance [pinot]
Posted by "klsince (via GitHub)" <gi...@apache.org>.
klsince commented on code in PR #11740:
URL: https://github.com/apache/pinot/pull/11740#discussion_r1347845423
##########
pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotTableRestletResource.java:
##########
@@ -615,11 +615,14 @@ public RebalanceResult rebalance(
@QueryParam("externalViewCheckIntervalInMs") long externalViewCheckIntervalInMs,
@ApiParam(value = "How long to wait till external view converges with ideal states") @DefaultValue("3600000")
@QueryParam("externalViewStabilizationTimeoutInMs") long externalViewStabilizationTimeoutInMs,
+ @ApiParam(value = "How often to make a status update (i.e. heartbeat)") @DefaultValue("300000")
Review Comment:
good catch, will add. I thought the rebalanceStatus API would simply return the whole jobMetadata which will include the retry info.
--
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
Re: [PR] Add support to retry failed table rebalance [pinot]
Posted by "Jackie-Jiang (via GitHub)" <gi...@apache.org>.
Jackie-Jiang commented on code in PR #11740:
URL: https://github.com/apache/pinot/pull/11740#discussion_r1351001671
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/rebalance/TableRebalanceRetryConfig.java:
##########
@@ -0,0 +1,74 @@
+/**
+ * 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;
+
+/**
+ * Track the job configs and retry numbers as part of the job ZK metadata for future retry.
+ */
+public class TableRebalanceRetryConfig {
Review Comment:
This is more like a context for rebalance, so suggest renaming it to `TableRebalanceContext`. Also, suggest moving `jobId` from `RebalanceConfig` into this class because that is also context
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/rebalance/RebalanceConfig.java:
##########
@@ -84,6 +84,24 @@ public class RebalanceConfig {
@ApiModelProperty(example = "false")
private boolean _updateTargetTier = false;
+ // Update job status every this interval as heartbeat, to indicate the job is still actively running.
+ @JsonProperty("heartbeatIntervalInMs")
+ @ApiModelProperty(example = "300000")
+ private long _heartbeatIntervalInMs = 300000L;
+
+ // The job is considered as failed if not updating its status by this timeout, even though it's IN_PROGRESS status.
+ @JsonProperty("heartbeatTimeoutInMs")
+ @ApiModelProperty(example = "3600000")
+ private long _heartbeatTimeoutInMs = 3600000L;
+
+ @JsonProperty("maxRetry")
+ @ApiModelProperty(example = "3")
+ private int _maxRetry = 3;
+
+ @JsonProperty("retryInitialDelayInMs")
+ @ApiModelProperty(example = "300000")
+ private long _retryInitialDelayInMs = 300000L;
Review Comment:
Not sure if we need this delay, or the exponential backoff. Rebalance task itself won't change ideal state before external view converges, thus no more messages will be sent before all messages for this table is processed.
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/rebalance/TableRebalanceRetryConfig.java:
##########
@@ -0,0 +1,74 @@
+/**
+ * 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;
+
+/**
+ * Track the job configs and retry numbers as part of the job ZK metadata for future retry.
+ */
+public class TableRebalanceRetryConfig {
+ private String _originalJobId;
+ private int _retryNum;
Review Comment:
For retry related config, we usually use `maxAttempts` and `attemptId`
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/rebalance/RebalanceChecker.java:
##########
@@ -0,0 +1,279 @@
+/**
+ * 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 com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Preconditions;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ExecutorService;
+import org.apache.commons.lang3.RandomUtils;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.commons.lang3.tuple.Pair;
+import org.apache.pinot.common.metadata.ZKMetadataProvider;
+import org.apache.pinot.common.metadata.controllerjob.ControllerJobType;
+import org.apache.pinot.common.metrics.ControllerMeter;
+import org.apache.pinot.common.metrics.ControllerMetrics;
+import org.apache.pinot.controller.ControllerConf;
+import org.apache.pinot.controller.LeadControllerManager;
+import org.apache.pinot.controller.helix.core.PinotHelixResourceManager;
+import org.apache.pinot.controller.helix.core.periodictask.ControllerPeriodicTask;
+import org.apache.pinot.spi.config.table.TableConfig;
+import org.apache.pinot.spi.utils.CommonConstants;
+import org.apache.pinot.spi.utils.JsonUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+/**
+ * Periodic task to check whether a user triggered rebalance job is completed or not, and retry if failed. The retry
+ * job is started with the same rebalance configs provided by the user and does best effort to stop the other jobs
+ * for the same table. This task can be configured to just check failures and report metrics, and not to do retry.
+ */
+public class RebalanceChecker extends ControllerPeriodicTask<Void> {
+ private static final Logger LOGGER = LoggerFactory.getLogger(RebalanceChecker.class);
+ private static final double RETRY_DELAY_SCALE_FACTOR = 2.0;
+ private final boolean _checkOnly;
+ private final ExecutorService _executorService;
+
+ public RebalanceChecker(PinotHelixResourceManager pinotHelixResourceManager,
+ LeadControllerManager leadControllerManager, ControllerConf config, ControllerMetrics controllerMetrics,
+ ExecutorService executorService) {
+ super(RebalanceChecker.class.getSimpleName(), config.getRebalanceCheckerFrequencyInSeconds(),
+ config.getRebalanceCheckerInitialDelayInSeconds(), pinotHelixResourceManager, leadControllerManager,
+ controllerMetrics);
+ _checkOnly = config.isRebalanceCheckerCheckOnly();
+ _executorService = executorService;
+ }
+
+ @Override
+ protected void processTable(String tableNameWithType) {
+ _executorService.submit(() -> {
Review Comment:
This will create one thread per table, and cause lots of ZK accesses. I wouldn't parallel them, but only schedule the needed retries into separate threads.
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/rebalance/RebalanceConfig.java:
##########
@@ -84,6 +84,24 @@ public class RebalanceConfig {
@ApiModelProperty(example = "false")
private boolean _updateTargetTier = false;
+ // Update job status every this interval as heartbeat, to indicate the job is still actively running.
+ @JsonProperty("heartbeatIntervalInMs")
+ @ApiModelProperty(example = "300000")
+ private long _heartbeatIntervalInMs = 300000L;
+
+ // The job is considered as failed if not updating its status by this timeout, even though it's IN_PROGRESS status.
+ @JsonProperty("heartbeatTimeoutInMs")
+ @ApiModelProperty(example = "3600000")
+ private long _heartbeatTimeoutInMs = 3600000L;
+
+ @JsonProperty("maxRetry")
+ @ApiModelProperty(example = "3")
+ private int _maxRetry = 3;
Review Comment:
Suggest changing it to `_maxAttempts`
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/rebalance/RebalanceChecker.java:
##########
@@ -0,0 +1,279 @@
+/**
+ * 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 com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Preconditions;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ExecutorService;
+import org.apache.commons.lang3.RandomUtils;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.commons.lang3.tuple.Pair;
+import org.apache.pinot.common.metadata.ZKMetadataProvider;
+import org.apache.pinot.common.metadata.controllerjob.ControllerJobType;
+import org.apache.pinot.common.metrics.ControllerMeter;
+import org.apache.pinot.common.metrics.ControllerMetrics;
+import org.apache.pinot.controller.ControllerConf;
+import org.apache.pinot.controller.LeadControllerManager;
+import org.apache.pinot.controller.helix.core.PinotHelixResourceManager;
+import org.apache.pinot.controller.helix.core.periodictask.ControllerPeriodicTask;
+import org.apache.pinot.spi.config.table.TableConfig;
+import org.apache.pinot.spi.utils.CommonConstants;
+import org.apache.pinot.spi.utils.JsonUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+/**
+ * Periodic task to check whether a user triggered rebalance job is completed or not, and retry if failed. The retry
+ * job is started with the same rebalance configs provided by the user and does best effort to stop the other jobs
+ * for the same table. This task can be configured to just check failures and report metrics, and not to do retry.
+ */
+public class RebalanceChecker extends ControllerPeriodicTask<Void> {
+ private static final Logger LOGGER = LoggerFactory.getLogger(RebalanceChecker.class);
+ private static final double RETRY_DELAY_SCALE_FACTOR = 2.0;
+ private final boolean _checkOnly;
+ private final ExecutorService _executorService;
+
+ public RebalanceChecker(PinotHelixResourceManager pinotHelixResourceManager,
+ LeadControllerManager leadControllerManager, ControllerConf config, ControllerMetrics controllerMetrics,
+ ExecutorService executorService) {
+ super(RebalanceChecker.class.getSimpleName(), config.getRebalanceCheckerFrequencyInSeconds(),
+ config.getRebalanceCheckerInitialDelayInSeconds(), pinotHelixResourceManager, leadControllerManager,
+ controllerMetrics);
+ _checkOnly = config.isRebalanceCheckerCheckOnly();
+ _executorService = executorService;
+ }
+
+ @Override
+ protected void processTable(String tableNameWithType) {
+ _executorService.submit(() -> {
+ try {
+ LOGGER.info("Start to retry rebalance for table: {}", tableNameWithType);
+ TableConfig tableConfig = _pinotHelixResourceManager.getTableConfig(tableNameWithType);
+ Preconditions.checkState(tableConfig != null, "Failed to find table config for table: {}", tableNameWithType);
+ retryRebalanceTable(tableNameWithType, tableConfig);
+ } catch (Throwable t) {
+ LOGGER.error("Failed to retry rebalance for table: {}", tableNameWithType, t);
+ }
+ });
+ }
+
+ @VisibleForTesting
+ void retryRebalanceTable(String tableNameWithType, TableConfig tableConfig)
+ throws Exception {
+ // Get all rebalance jobs as tracked in ZK for this table to check if there is any failure to retry.
+ Map<String, Map<String, String>> allJobMetadata = _pinotHelixResourceManager.getAllJobsForTable(tableNameWithType,
Review Comment:
We don't want to do this per table. Instead, we may read all rebalance job metadata once, and then process all of them together. You may directly override `processTables()` which process all tables together
##########
pinot-controller/src/main/java/org/apache/pinot/controller/ControllerConf.java:
##########
@@ -199,6 +199,9 @@ public static class ControllerPeriodicTasksConf {
// RealtimeSegmentRelocator has been rebranded as SegmentRelocator
public static final String DEPRECATED_REALTIME_SEGMENT_RELOCATION_INITIAL_DELAY_IN_SECONDS =
"controller.realtimeSegmentRelocation.initialDelayInSeconds";
+ public static final String REBALANCE_CHECKER_INITIAL_DELAY_IN_SECONDS =
+ "controller.rebalanceChecker.initialDelayInSeconds";
+ public static final String REBALANCE_CHECKER_CHECK_ONLY = "controller.rebalanceChecker.checkOnly";
Review Comment:
This config will void the retry from the rebalance call, so I definitely won't make it the default. IMO we can just remove it, and if by any change rebalance checker has a bug, we can turn it completely off
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/rebalance/TableRebalancer.java:
##########
@@ -137,6 +139,23 @@ public static String createUniqueRebalanceJobIdentifier() {
}
public RebalanceResult rebalance(TableConfig tableConfig, RebalanceConfig rebalanceConfig) {
+ long startTime = System.currentTimeMillis();
+ String tableNameWithType = tableConfig.getTableName();
+ String status = "ABORTED";
Review Comment:
What does `ABORTED` stand for? When would it happen?
Currently seems it can occur only when exception is somehow not caught. Should we call it `ERRORED` and make it a valid `Status` enum?
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/rebalance/ZkBasedTableRebalanceObserver.java:
##########
@@ -126,30 +143,73 @@ public int getNumUpdatesToZk() {
return _numUpdatesToZk;
}
+ @VisibleForTesting
+ TableRebalanceRetryConfig getTableRebalanceJobRetryConfig() {
+ return _tableRebalanceJobRetryConfig;
+ }
+
private void trackStatsInZk() {
+ Map<String, String> jobMetadata =
+ createJobMetadata(_tableNameWithType, _rebalanceJobId, _tableRebalanceProgressStats,
+ _tableRebalanceJobRetryConfig);
+ _pinotHelixResourceManager.addControllerJobToZK(_rebalanceJobId, jobMetadata,
+ ZKMetadataProvider.constructPropertyStorePathForControllerJob(ControllerJobType.TABLE_REBALANCE),
+ prevJobMetadata -> {
+ // Abort the job when we're sure it has failed, otherwise continue to update the status.
+ if (prevJobMetadata == null) {
+ return true;
+ }
+ String prevStatsInStr = prevJobMetadata.get(RebalanceJobConstants.JOB_METADATA_KEY_REBALANCE_PROGRESS_STATS);
+ TableRebalanceProgressStats prevStats;
+ try {
+ prevStats = JsonUtils.stringToObject(prevStatsInStr, TableRebalanceProgressStats.class);
+ } catch (JsonProcessingException ignore) {
+ return true;
+ }
+ boolean jobFailed =
+ prevStats != null && RebalanceResult.Status.FAILED.toString().equals(prevStats.getStatus());
Review Comment:
Now I see how this abort is wired. I feel the abstraction is not correct though because basically we allow an observer to abort the execution by throwing an runtime exception. This is not safe because we cannot differentiate this abort vs other exceptions.
One option to make it cleaner is to add an API `isAborted()` into the observer, and call it in the `TableRebalancer`
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/rebalance/ZkBasedTableRebalanceObserver.java:
##########
@@ -98,10 +117,9 @@ private void updateOnStart(Map<String, Map<String, String>> currentState,
@Override
public void onSuccess(String msg) {
- Preconditions.checkState(_tableRebalanceProgressStats.getStatus() != RebalanceResult.Status.DONE.toString(),
+ Preconditions.checkState(!RebalanceResult.Status.DONE.toString().equals(_tableRebalanceProgressStats.getStatus()),
Review Comment:
Good catch! IMO a better fix is to directly put `Status` instead of `String` in the `_tableRebalanceProgressStats`
--
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
Re: [PR] Add support to retry failed table rebalance [pinot]
Posted by "klsince (via GitHub)" <gi...@apache.org>.
klsince commented on code in PR #11740:
URL: https://github.com/apache/pinot/pull/11740#discussion_r1361278366
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/rebalance/TableRebalanceAttemptContext.java:
##########
@@ -0,0 +1,108 @@
+/**
+ * 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 com.google.common.annotations.VisibleForTesting;
+
+
+/**
+ * Track the job configs and attempt numbers as part of the job ZK metadata for future retry.
+ */
+public class TableRebalanceAttemptContext {
Review Comment:
in fact, `Attempt` was meant to include both original and retry runs. but sgtm to name this more generic.
--
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
Re: [PR] Add support to retry failed table rebalance [pinot]
Posted by "klsince (via GitHub)" <gi...@apache.org>.
klsince commented on code in PR #11740:
URL: https://github.com/apache/pinot/pull/11740#discussion_r1355843815
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/rebalance/TableRebalanceRetryConfig.java:
##########
@@ -0,0 +1,74 @@
+/**
+ * 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;
+
+/**
+ * Track the job configs and retry numbers as part of the job ZK metadata for future retry.
+ */
+public class TableRebalanceRetryConfig {
Review Comment:
sgtm, renamed to TableRebalanceAttemptContext, as it'll have the attemptId field.
But moving `jobId` out of `RebalanceConfig` would require changes on PinotHelixResourceManager.rebalanceTable() and TableRebalancer.rebalance() to accept `jobId` along with `RebalanceConfig`. If you feel strong about this, I can open a PR to refactor on this firstly.
--
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
Re: [PR] Add support to retry failed table rebalance [pinot]
Posted by "klsince (via GitHub)" <gi...@apache.org>.
klsince commented on PR #11740:
URL: https://github.com/apache/pinot/pull/11740#issuecomment-1760593159
> Can we provide an rest API to abort the current rebalance and stop the following retries?
Good idea. This would need a new ABORTED status, different from FAILED status. Aborted jobs are not considered for retry. The API might look like `DELETE /rebalance/{tableName}/{jobId}` where jobId is optional. When jobId is omit, all rebalance jobs are aborted for the given table.
--
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
Re: [PR] Add support to retry failed table rebalance [pinot]
Posted by "klsince (via GitHub)" <gi...@apache.org>.
klsince commented on code in PR #11740:
URL: https://github.com/apache/pinot/pull/11740#discussion_r1353180549
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/rebalance/RebalanceConfig.java:
##########
@@ -84,6 +84,24 @@ public class RebalanceConfig {
@ApiModelProperty(example = "false")
private boolean _updateTargetTier = false;
+ // Update job status every this interval as heartbeat, to indicate the job is still actively running.
+ @JsonProperty("heartbeatIntervalInMs")
+ @ApiModelProperty(example = "300000")
+ private long _heartbeatIntervalInMs = 300000L;
+
+ // The job is considered as failed if not updating its status by this timeout, even though it's IN_PROGRESS status.
+ @JsonProperty("heartbeatTimeoutInMs")
+ @ApiModelProperty(example = "3600000")
+ private long _heartbeatTimeoutInMs = 3600000L;
+
+ @JsonProperty("maxRetry")
+ @ApiModelProperty(example = "3")
+ private int _maxRetry = 3;
+
+ @JsonProperty("retryInitialDelayInMs")
+ @ApiModelProperty(example = "300000")
+ private long _retryInitialDelayInMs = 300000L;
Review Comment:
I think this is still useful,to avoid herd effect particularly for clusters with many tables. If many tables are doing rebalance but all failed like because controllers got restarted, then all rebalances would get retried all at once when RebalanceChecker task runs next time. If wanted, setting this to 0 can disable the backoff mechanism.
--
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
Re: [PR] Add support to retry failed table rebalance [pinot]
Posted by "klsince (via GitHub)" <gi...@apache.org>.
klsince commented on code in PR #11740:
URL: https://github.com/apache/pinot/pull/11740#discussion_r1364309394
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/rebalance/RebalanceChecker.java:
##########
@@ -0,0 +1,295 @@
+/**
+ * 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 com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Preconditions;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ExecutorService;
+import org.apache.commons.lang3.RandomUtils;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.commons.lang3.tuple.Pair;
+import org.apache.pinot.common.metadata.controllerjob.ControllerJobType;
+import org.apache.pinot.common.metrics.ControllerMeter;
+import org.apache.pinot.common.metrics.ControllerMetrics;
+import org.apache.pinot.controller.ControllerConf;
+import org.apache.pinot.controller.LeadControllerManager;
+import org.apache.pinot.controller.helix.core.PinotHelixResourceManager;
+import org.apache.pinot.controller.helix.core.periodictask.ControllerPeriodicTask;
+import org.apache.pinot.spi.config.table.TableConfig;
+import org.apache.pinot.spi.utils.CommonConstants;
+import org.apache.pinot.spi.utils.JsonUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+/**
+ * Periodic task to check whether a user triggered rebalance job is completed or not, and retry if failed. The retry
+ * job is started with the same rebalance configs provided by the user and does best effort to stop the other jobs
+ * for the same table. This task can be configured to just check failures and report metrics, and not to do retry.
+ */
+public class RebalanceChecker extends ControllerPeriodicTask<Void> {
+ private static final Logger LOGGER = LoggerFactory.getLogger(RebalanceChecker.class);
+ private static final double RETRY_DELAY_SCALE_FACTOR = 2.0;
+ private final ExecutorService _executorService;
+
+ public RebalanceChecker(PinotHelixResourceManager pinotHelixResourceManager,
+ LeadControllerManager leadControllerManager, ControllerConf config, ControllerMetrics controllerMetrics,
+ ExecutorService executorService) {
+ super(RebalanceChecker.class.getSimpleName(), config.getRebalanceCheckerFrequencyInSeconds(),
+ config.getRebalanceCheckerInitialDelayInSeconds(), pinotHelixResourceManager, leadControllerManager,
+ controllerMetrics);
+ _executorService = executorService;
+ }
+
+ @Override
+ protected void processTable(String tableNameWithType) {
+ try {
+ LOGGER.info("Start to retry rebalance for table: {}", tableNameWithType);
+ // Get all rebalance jobs as tracked in ZK for this table to check if there is any failure to retry.
+ Map<String, Map<String, String>> allJobMetadata = _pinotHelixResourceManager.getAllJobsForTable(tableNameWithType,
Review Comment:
interesting, how this behavior is configured? but anyway, I've updated the PR to read ZK once to get job metadata for all the given tables.
--
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
Re: [PR] Add support to retry failed table rebalance [pinot]
Posted by "klsince (via GitHub)" <gi...@apache.org>.
klsince commented on code in PR #11740:
URL: https://github.com/apache/pinot/pull/11740#discussion_r1361217787
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/rebalance/RebalanceChecker.java:
##########
@@ -0,0 +1,330 @@
+/**
+ * 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 com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Preconditions;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.Set;
+import java.util.concurrent.ExecutorService;
+import org.apache.commons.lang3.RandomUtils;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.commons.lang3.tuple.Pair;
+import org.apache.pinot.common.metadata.ZKMetadataProvider;
+import org.apache.pinot.common.metadata.controllerjob.ControllerJobType;
+import org.apache.pinot.common.metrics.ControllerGauge;
+import org.apache.pinot.common.metrics.ControllerMeter;
+import org.apache.pinot.common.metrics.ControllerMetrics;
+import org.apache.pinot.controller.ControllerConf;
+import org.apache.pinot.controller.LeadControllerManager;
+import org.apache.pinot.controller.helix.core.PinotHelixResourceManager;
+import org.apache.pinot.controller.helix.core.periodictask.ControllerPeriodicTask;
+import org.apache.pinot.spi.config.table.TableConfig;
+import org.apache.pinot.spi.utils.CommonConstants;
+import org.apache.pinot.spi.utils.JsonUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+/**
+ * Periodic task to check whether a user triggered rebalance job is completed or not, and retry if failed. The retry
+ * job is started with the same rebalance configs provided by the user and does best effort to stop the other jobs
+ * for the same table. This task can be configured to just check failures and report metrics, and not to do retry.
+ */
+public class RebalanceChecker extends ControllerPeriodicTask<Void> {
+ private static final Logger LOGGER = LoggerFactory.getLogger(RebalanceChecker.class);
+ private static final double RETRY_DELAY_SCALE_FACTOR = 2.0;
+ private final ExecutorService _executorService;
+
+ public RebalanceChecker(PinotHelixResourceManager pinotHelixResourceManager,
+ LeadControllerManager leadControllerManager, ControllerConf config, ControllerMetrics controllerMetrics,
+ ExecutorService executorService) {
+ super(RebalanceChecker.class.getSimpleName(), config.getRebalanceCheckerFrequencyInSeconds(),
+ config.getRebalanceCheckerInitialDelayInSeconds(), pinotHelixResourceManager, leadControllerManager,
+ controllerMetrics);
+ _executorService = executorService;
+ }
+
+ protected void processTables(List<String> tableNamesWithType, Properties periodicTaskProperties) {
+ int numTables = tableNamesWithType.size();
+ LOGGER.info("Processing {} tables in task: {}", numTables, _taskName);
+ int numTablesProcessed = 0;
+ for (String tableNameWithType : tableNamesWithType) {
+ if (!isStarted()) {
+ LOGGER.info("Task: {} is stopped, early terminate the task", _taskName);
+ break;
+ }
+ try {
+ processTable(tableNameWithType);
Review Comment:
> ... job metadata 1000 times
there is only one ZNode to keep all rebalance job metadata from all tables and there is cache in zkClient, so it's not 1000 reads against ZK. But I should delay access to tableConfig after the table is decided for another retry.
btw, I just removed the override of processTables() method as I didn't really change its default implementation in this PR.
--
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
Re: [PR] Add support to retry failed table rebalance [pinot]
Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #11740:
URL: https://github.com/apache/pinot/pull/11740#issuecomment-1747731003
## [Codecov](https://app.codecov.io/gh/apache/pinot/pull/11740?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
> Merging [#11740](https://app.codecov.io/gh/apache/pinot/pull/11740?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) (8ca19cf) into [master](https://app.codecov.io/gh/apache/pinot/commit/c96221f697ab9347a1e5dbf145a9d6d660c53d62?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) (c96221f) will **decrease** coverage by `63.12%`.
> Report is 1 commits behind head on master.
> The diff coverage is `0.00%`.
```diff
@@ Coverage Diff @@
## master #11740 +/- ##
=============================================
- Coverage 63.11% 0.00% -63.12%
=============================================
Files 2342 2268 -74
Lines 125899 122403 -3496
Branches 19364 18841 -523
=============================================
- Hits 79464 0 -79464
- Misses 40780 122403 +81623
+ Partials 5655 0 -5655
```
| [Flag](https://app.codecov.io/gh/apache/pinot/pull/11740/flags?src=pr&el=flags&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [integration](https://app.codecov.io/gh/apache/pinot/pull/11740/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `0.00% <0.00%> (-0.01%)` | :arrow_down: |
| [integration1](https://app.codecov.io/gh/apache/pinot/pull/11740/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
| [integration2](https://app.codecov.io/gh/apache/pinot/pull/11740/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `0.00% <0.00%> (ø)` | |
| [java-11](https://app.codecov.io/gh/apache/pinot/pull/11740/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `0.00% <0.00%> (-63.06%)` | :arrow_down: |
| [java-17](https://app.codecov.io/gh/apache/pinot/pull/11740/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
| [java-20](https://app.codecov.io/gh/apache/pinot/pull/11740/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
| [temurin](https://app.codecov.io/gh/apache/pinot/pull/11740/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `0.00% <0.00%> (-63.12%)` | :arrow_down: |
| [unittests](https://app.codecov.io/gh/apache/pinot/pull/11740/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
| [unittests1](https://app.codecov.io/gh/apache/pinot/pull/11740/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
| [unittests2](https://app.codecov.io/gh/apache/pinot/pull/11740/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#carryforward-flags-in-the-pull-request-comment) to find out more.
| [Files](https://app.codecov.io/gh/apache/pinot/pull/11740?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [...g/apache/pinot/common/metrics/ControllerTimer.java](https://app.codecov.io/gh/apache/pinot/pull/11740?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cGlub3QtY29tbW9uL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9jb21tb24vbWV0cmljcy9Db250cm9sbGVyVGltZXIuamF2YQ==) | `0.00% <0.00%> (ø)` | |
| [...apache/pinot/controller/BaseControllerStarter.java](https://app.codecov.io/gh/apache/pinot/pull/11740?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cGlub3QtY29udHJvbGxlci9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcGlub3QvY29udHJvbGxlci9CYXNlQ29udHJvbGxlclN0YXJ0ZXIuamF2YQ==) | `0.00% <0.00%> (-78.17%)` | :arrow_down: |
| [...g/apache/pinot/common/metrics/ControllerMeter.java](https://app.codecov.io/gh/apache/pinot/pull/11740?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cGlub3QtY29tbW9uL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9jb21tb24vbWV0cmljcy9Db250cm9sbGVyTWV0ZXIuamF2YQ==) | `0.00% <0.00%> (ø)` | |
| [...oller/api/resources/PinotTableRestletResource.java](https://app.codecov.io/gh/apache/pinot/pull/11740?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cGlub3QtY29udHJvbGxlci9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcGlub3QvY29udHJvbGxlci9hcGkvcmVzb3VyY2VzL1Bpbm90VGFibGVSZXN0bGV0UmVzb3VyY2UuamF2YQ==) | `0.00% <0.00%> (-45.75%)` | :arrow_down: |
| [...va/org/apache/pinot/controller/ControllerConf.java](https://app.codecov.io/gh/apache/pinot/pull/11740?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cGlub3QtY29udHJvbGxlci9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcGlub3QvY29udHJvbGxlci9Db250cm9sbGVyQ29uZi5qYXZh) | `0.00% <0.00%> (-58.05%)` | :arrow_down: |
| [...ntroller/helix/core/rebalance/TableRebalancer.java](https://app.codecov.io/gh/apache/pinot/pull/11740?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cGlub3QtY29udHJvbGxlci9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcGlub3QvY29udHJvbGxlci9oZWxpeC9jb3JlL3JlYmFsYW5jZS9UYWJsZVJlYmFsYW5jZXIuamF2YQ==) | `0.00% <0.00%> (-68.59%)` | :arrow_down: |
| [...ntroller/helix/core/rebalance/RebalanceConfig.java](https://app.codecov.io/gh/apache/pinot/pull/11740?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cGlub3QtY29udHJvbGxlci9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcGlub3QvY29udHJvbGxlci9oZWxpeC9jb3JlL3JlYmFsYW5jZS9SZWJhbGFuY2VDb25maWcuamF2YQ==) | `0.00% <0.00%> (-100.00%)` | :arrow_down: |
| [...elix/core/rebalance/TableRebalanceRetryConfig.java](https://app.codecov.io/gh/apache/pinot/pull/11740?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cGlub3QtY29udHJvbGxlci9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcGlub3QvY29udHJvbGxlci9oZWxpeC9jb3JlL3JlYmFsYW5jZS9UYWJsZVJlYmFsYW5jZVJldHJ5Q29uZmlnLmphdmE=) | `0.00% <0.00%> (ø)` | |
| [...ntroller/helix/core/PinotHelixResourceManager.java](https://app.codecov.io/gh/apache/pinot/pull/11740?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cGlub3QtY29udHJvbGxlci9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcGlub3QvY29udHJvbGxlci9oZWxpeC9jb3JlL1Bpbm90SGVsaXhSZXNvdXJjZU1hbmFnZXIuamF2YQ==) | `0.00% <0.00%> (-63.81%)` | :arrow_down: |
| [.../core/rebalance/ZkBasedTableRebalanceObserver.java](https://app.codecov.io/gh/apache/pinot/pull/11740?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cGlub3QtY29udHJvbGxlci9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcGlub3QvY29udHJvbGxlci9oZWxpeC9jb3JlL3JlYmFsYW5jZS9aa0Jhc2VkVGFibGVSZWJhbGFuY2VPYnNlcnZlci5qYXZh) | `0.00% <0.00%> (-80.44%)` | :arrow_down: |
| ... and [1 more](https://app.codecov.io/gh/apache/pinot/pull/11740?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | |
... and [1987 files with indirect coverage changes](https://app.codecov.io/gh/apache/pinot/pull/11740/indirect-changes?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)
:mega: We’re building smart automated test selection to slash your CI/CD build times. [Learn more](https://about.codecov.io/iterative-testing/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)
--
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
Re: [PR] Add support to retry failed table rebalance [pinot]
Posted by "klsince (via GitHub)" <gi...@apache.org>.
klsince commented on code in PR #11740:
URL: https://github.com/apache/pinot/pull/11740#discussion_r1361220254
##########
pinot-core/src/main/java/org/apache/pinot/core/auth/Actions.java:
##########
@@ -89,6 +89,7 @@ public static class Cluster {
// Action names for table
public static class Table {
+ public static final String ABORT_REBALANCE = "AbortRebalance";
Review Comment:
sgtm, particularly with the CANCELLED status.
--
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
Re: [PR] Add support to retry failed table rebalance [pinot]
Posted by "klsince (via GitHub)" <gi...@apache.org>.
klsince commented on code in PR #11740:
URL: https://github.com/apache/pinot/pull/11740#discussion_r1361252175
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManager.java:
##########
@@ -2063,28 +2070,76 @@ public boolean addNewReloadAllSegmentsJob(String tableNameWithType, String jobId
ZKMetadataProvider.constructPropertyStorePathForControllerJob(ControllerJobType.RELOAD_SEGMENT));
}
+ /**
+ * Adds a new job metadata for controller job like table rebalance or reload into ZK
+ * @param jobId job's UUID
+ * @param jobMetadata the job metadata
+ * @param jobResourcePath zkPath to add the new job metadata
+ * @return boolean representing success / failure of the ZK write step
+ */
public boolean addControllerJobToZK(String jobId, Map<String, String> jobMetadata, String jobResourcePath) {
Review Comment:
sgtm. will refactor this here.
--
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
Re: [PR] Add support to retry failed table rebalance [pinot]
Posted by "Jackie-Jiang (via GitHub)" <gi...@apache.org>.
Jackie-Jiang commented on code in PR #11740:
URL: https://github.com/apache/pinot/pull/11740#discussion_r1362962952
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManager.java:
##########
@@ -2063,28 +2070,76 @@ public boolean addNewReloadAllSegmentsJob(String tableNameWithType, String jobId
ZKMetadataProvider.constructPropertyStorePathForControllerJob(ControllerJobType.RELOAD_SEGMENT));
}
+ /**
+ * Adds a new job metadata for controller job like table rebalance or reload into ZK
+ * @param jobId job's UUID
+ * @param jobMetadata the job metadata
+ * @param jobResourcePath zkPath to add the new job metadata
+ * @return boolean representing success / failure of the ZK write step
+ */
public boolean addControllerJobToZK(String jobId, Map<String, String> jobMetadata, String jobResourcePath) {
+ return addControllerJobToZK(jobId, jobMetadata, jobResourcePath, prev -> true);
+ }
+
+ /**
+ * Adds a new job metadata for controller job like table rebalance or reload into ZK
+ * @param jobId job's UUID
+ * @param jobMetadata the job metadata
+ * @param jobResourcePath zkPath to add the new job metadata
+ * @param prevJobMetadataChecker to check the previous job metadata before adding new one
+ * @return boolean representing success / failure of the ZK write step
+ */
+ public boolean addControllerJobToZK(String jobId, Map<String, String> jobMetadata, String jobResourcePath,
+ Predicate<Map<String, String>> prevJobMetadataChecker) {
Preconditions.checkState(jobMetadata.get(CommonConstants.ControllerJob.SUBMISSION_TIME_MS) != null,
"Submission Time in JobMetadata record not set. Cannot expire these records");
Stat stat = new Stat();
- ZNRecord tableJobsZnRecord = _propertyStore.get(jobResourcePath, stat, AccessOption.PERSISTENT);
- if (tableJobsZnRecord != null) {
- Map<String, Map<String, String>> tasks = tableJobsZnRecord.getMapFields();
- tasks.put(jobId, jobMetadata);
- if (tasks.size() > CommonConstants.ControllerJob.MAXIMUM_CONTROLLER_JOBS_IN_ZK) {
- tasks = tasks.entrySet().stream().sorted((v1, v2) -> Long.compare(
+ ZNRecord jobsZnRecord = _propertyStore.get(jobResourcePath, stat, AccessOption.PERSISTENT);
+ if (jobsZnRecord != null) {
+ Map<String, Map<String, String>> jobMetadataMap = jobsZnRecord.getMapFields();
+ Map<String, String> prevJobMetadata = jobMetadataMap.get(jobId);
+ if (!prevJobMetadataChecker.test(prevJobMetadata)) {
+ return false;
+ }
+ jobMetadataMap.put(jobId, jobMetadata);
+ if (jobMetadataMap.size() > CommonConstants.ControllerJob.MAXIMUM_CONTROLLER_JOBS_IN_ZK) {
+ jobMetadataMap = jobMetadataMap.entrySet().stream().sorted((v1, v2) -> Long.compare(
Long.parseLong(v2.getValue().get(CommonConstants.ControllerJob.SUBMISSION_TIME_MS)),
Long.parseLong(v1.getValue().get(CommonConstants.ControllerJob.SUBMISSION_TIME_MS))))
.collect(Collectors.toList()).subList(0, CommonConstants.ControllerJob.MAXIMUM_CONTROLLER_JOBS_IN_ZK)
.stream().collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
}
- tableJobsZnRecord.setMapFields(tasks);
- return _propertyStore.set(jobResourcePath, tableJobsZnRecord, stat.getVersion(), AccessOption.PERSISTENT);
+ jobsZnRecord.setMapFields(jobMetadataMap);
+ return _propertyStore.set(jobResourcePath, jobsZnRecord, stat.getVersion(), AccessOption.PERSISTENT);
} else {
- tableJobsZnRecord = new ZNRecord(jobResourcePath);
- tableJobsZnRecord.setMapField(jobId, jobMetadata);
- return _propertyStore.set(jobResourcePath, tableJobsZnRecord, AccessOption.PERSISTENT);
+ jobsZnRecord = new ZNRecord(jobResourcePath);
+ jobsZnRecord.setMapField(jobId, jobMetadata);
+ return _propertyStore.set(jobResourcePath, jobsZnRecord, AccessOption.PERSISTENT);
+ }
+ }
+
+ /**
+ * Update existing job metadata belong to the table
+ * @param tableNameWithType whose job metadata to be updated
+ * @param jobResourcePath zkPath to look for the job metadata for the table
+ * @param updater to modify the job metadata in place
+ * @return boolean representing success / failure of the ZK write step
+ */
+ public boolean updateAllJobsForTable(String tableNameWithType, String jobResourcePath,
Review Comment:
nvm, didn't realize we need to filter on table name first
--
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
Re: [PR] Add support to retry failed table rebalance [pinot]
Posted by "klsince (via GitHub)" <gi...@apache.org>.
klsince commented on code in PR #11740:
URL: https://github.com/apache/pinot/pull/11740#discussion_r1347844036
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/rebalance/RebalanceChecker.java:
##########
@@ -0,0 +1,258 @@
+/**
+ * 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 com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Preconditions;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ExecutorService;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.commons.lang3.tuple.Pair;
+import org.apache.pinot.common.metadata.ZKMetadataProvider;
+import org.apache.pinot.common.metadata.controllerjob.ControllerJobType;
+import org.apache.pinot.common.metrics.ControllerMeter;
+import org.apache.pinot.common.metrics.ControllerMetrics;
+import org.apache.pinot.controller.ControllerConf;
+import org.apache.pinot.controller.LeadControllerManager;
+import org.apache.pinot.controller.helix.core.PinotHelixResourceManager;
+import org.apache.pinot.controller.helix.core.periodictask.ControllerPeriodicTask;
+import org.apache.pinot.spi.config.table.TableConfig;
+import org.apache.pinot.spi.utils.CommonConstants;
+import org.apache.pinot.spi.utils.JsonUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+/**
+ * Periodic task to check whether a user triggered rebalance job is completed or not, and retry if failed. The retry
+ * job is started with the same rebalance configs provided by the user and does best effort to stop the other jobs
+ * for the same table. This task can be configured to just check failures and report metrics, and not to do retry.
+ */
+public class RebalanceChecker extends ControllerPeriodicTask<Void> {
+ private static final Logger LOGGER = LoggerFactory.getLogger(RebalanceChecker.class);
+ private final boolean _checkOnly;
+ private final ExecutorService _executorService;
+
+ public RebalanceChecker(PinotHelixResourceManager pinotHelixResourceManager,
+ LeadControllerManager leadControllerManager, ControllerConf config, ControllerMetrics controllerMetrics,
+ ExecutorService executorService) {
+ super(RebalanceChecker.class.getSimpleName(), config.getRebalanceCheckerFrequencyInSeconds(),
+ config.getRebalanceCheckerInitialDelayInSeconds(), pinotHelixResourceManager, leadControllerManager,
+ controllerMetrics);
+ _checkOnly = config.isRebalanceCheckerCheckOnly();
+ _executorService = executorService;
+ }
+
+ @Override
+ protected void processTable(String tableNameWithType) {
+ _executorService.submit(() -> {
+ try {
+ TableConfig tableConfig = _pinotHelixResourceManager.getTableConfig(tableNameWithType);
+ Preconditions.checkState(tableConfig != null, "Failed to find table config for table: {}", tableNameWithType);
+ retryRebalanceTable(tableNameWithType, tableConfig);
+ } catch (Throwable t) {
+ LOGGER.error("Failed to retry rebalance for table: {}", tableNameWithType, t);
+ }
+ });
+ }
+
+ @VisibleForTesting
+ void retryRebalanceTable(String tableNameWithType, TableConfig tableConfig)
+ throws Exception {
+ // Get all rebalance jobs as tracked in ZK for this table to check if there is any failure to retry.
+ Map<String, Map<String, String>> allJobMetadata = _pinotHelixResourceManager.getAllJobsForTable(tableNameWithType,
+ Collections.singleton(ControllerJobType.TABLE_REBALANCE));
+ if (allJobMetadata.isEmpty()) {
+ LOGGER.info("No rebalance job has been triggered for table: {}. Skip retry", tableNameWithType);
+ return;
+ }
+ // Skip retry for the table if rebalance job is still running or has completed, in specific:
+ // 1) Skip retry if any rebalance job is actively running. Being actively running means the job is at IN_PROGRESS
+ // status, and has updated its status kept in ZK within the heartbeat timeout. It's possible that more than one
+ // rebalance jobs are running for the table, but that's fine with idempotent rebalance algorithm.
+ // 2) Skip retry if the most recently started rebalance job has completed with DONE or NO_OP. It's possible that
+ // jobs started earlier may be still running, but they are ignored here.
+ //
+ // Otherwise, we can get a list of failed rebalance jobs, i.e. those at FAILED status; or IN_PROGRESS status but
+ // haven't updated their status kept in ZK within the heartbeat timeout. For those candidate jobs to retry:
+ // 1) Firstly, group them by the original jobIds they retry for so that we can skip those exceeded maxRetry.
+ // 2) For the remaining jobs, we take the one started most recently and retry it with its original configs.
+ // 3) If configured, we can abort the other rebalance jobs for the table by setting their status to FAILED.
+ Map<String/*original jobId*/, Set<Pair<TableRebalanceRetryConfig/*original or retry jobId*/, Long
+ /*job startTime*/>>> candidateJobs = getCandidateJobs(tableNameWithType, allJobMetadata);
+ if (candidateJobs.isEmpty()) {
+ LOGGER.info("Found no failed rebalance jobs for table: {}. Skip retry", tableNameWithType);
+ return;
+ }
+ _controllerMetrics.addMeteredTableValue(tableNameWithType, ControllerMeter.TABLE_REBALANCE_FAILURE_DETECTED, 1L);
+ if (_checkOnly) {
+ LOGGER.info("Found failed rebalance jobs: {} for table: {}, but check only", candidateJobs.keySet(),
+ tableNameWithType);
+ return;
+ }
+ TableRebalanceRetryConfig retryConfig = getLatestJob(candidateJobs);
+ if (retryConfig == null) {
+ LOGGER.info("Rebalance is retried enough times for table: {}. Skip retry", tableNameWithType);
+ return;
+ }
+ tryStopExistingJobs(tableNameWithType, _pinotHelixResourceManager);
+ String newRetryJobId = TableRebalancer.createUniqueRebalanceJobIdentifier();
+ // Reset the JOB_ID of rebalanceConfig to be the new retry job ID.
+ RebalanceConfig rebalanceConfig = retryConfig.getConfig();
+ rebalanceConfig.setJobId(newRetryJobId);
+ ZkBasedTableRebalanceObserver observer = new ZkBasedTableRebalanceObserver(tableNameWithType, newRetryJobId,
+ TableRebalanceRetryConfig.forRetryRun(retryConfig.getOriginalJobId(), rebalanceConfig,
+ retryConfig.getRetryNum() + 1), _pinotHelixResourceManager);
+ LOGGER.info("Retry original rebalance job: {} for table: {} with new retry job: {}", retryConfig.getOriginalJobId(),
+ tableNameWithType, newRetryJobId);
+ _controllerMetrics.addMeteredTableValue(tableNameWithType, ControllerMeter.TABLE_REBALANCE_RETRY, 1L);
+ RebalanceResult result =
+ _pinotHelixResourceManager.rebalanceTable(tableNameWithType, tableConfig, newRetryJobId, rebalanceConfig,
+ observer);
+ LOGGER.info("Retry job: {} for table: {} is done with result: {}", newRetryJobId, tableNameWithType, result);
+ }
+
+ private static void tryStopExistingJobs(String tableNameWithType,
+ PinotHelixResourceManager pinotHelixResourceManager) {
+ pinotHelixResourceManager.updateAllJobsForTable(tableNameWithType,
+ ZKMetadataProvider.constructPropertyStorePathForControllerJob(ControllerJobType.TABLE_REBALANCE),
+ jobMetadata -> updateJobMetadata(tableNameWithType, jobMetadata));
+ }
+
+ private static void updateJobMetadata(String tableNameWithType, Map<String, String> jobMetadata) {
+ String jobId = jobMetadata.get(CommonConstants.ControllerJob.JOB_ID);
+ try {
+ String jobStatsInStr = jobMetadata.get(RebalanceJobConstants.JOB_METADATA_KEY_REBALANCE_PROGRESS_STATS);
+ TableRebalanceProgressStats jobStats = JsonUtils.stringToObject(jobStatsInStr, TableRebalanceProgressStats.class);
+ LOGGER.info("Set status of rebalance job: {} for table: {} from: {} to: FAILED", jobId, tableNameWithType,
+ jobStats.getStatus());
+ jobStats.setStatus(RebalanceResult.Status.FAILED.toString());
+ jobMetadata.put(RebalanceJobConstants.JOB_METADATA_KEY_REBALANCE_PROGRESS_STATS,
+ JsonUtils.objectToString(jobStats));
+ } catch (Exception e) {
+ LOGGER.error("Failed to update status of rebalance job: {} for table: {}", jobId, tableNameWithType, e);
+ }
+ }
+
+ @VisibleForTesting
+ static TableRebalanceRetryConfig getLatestJob(Map<String, Set<Pair<TableRebalanceRetryConfig, Long>>> candidateJobs) {
+ Pair<TableRebalanceRetryConfig, Long> candidateJobRun = null;
+ for (Map.Entry<String, Set<Pair<TableRebalanceRetryConfig, Long>>> entry : candidateJobs.entrySet()) {
+ // The job configs from all retry jobs are same, as the same set of job configs is used to do retry.
+ // The job metadata kept in ZK is cleaned by submission time order gradually, so we can't compare Set.size()
+ // against maxRetry, but check retryNum of each run to see if retries have exceeded limit.
+ Set<Pair<TableRebalanceRetryConfig, Long>> jobRuns = entry.getValue();
+ int maxRetry = jobRuns.iterator().next().getLeft().getConfig().getMaxRetry();
+ Pair<TableRebalanceRetryConfig, Long> latestJobRun = null;
+ for (Pair<TableRebalanceRetryConfig, Long> jobRun : jobRuns) {
+ if (jobRun.getLeft().getRetryNum() >= maxRetry) {
+ latestJobRun = null;
+ break;
+ }
+ if (latestJobRun == null || latestJobRun.getRight() < jobRun.getRight()) {
+ latestJobRun = jobRun;
+ }
+ }
+ if (latestJobRun == null) {
+ LOGGER.info("Rebalance job: {} had exceeded maxRetry: {}. Skip retry", entry.getKey(), maxRetry);
+ continue;
+ }
+ if (candidateJobRun == null || candidateJobRun.getRight() < latestJobRun.getRight()) {
+ candidateJobRun = latestJobRun;
+ }
+ }
+ return candidateJobRun == null ? null : candidateJobRun.getLeft();
+ }
+
+ @VisibleForTesting
+ static Map<String, Set<Pair<TableRebalanceRetryConfig, Long>>> getCandidateJobs(String tableNameWithType,
+ Map<String, Map<String, String>> allJobMetadata)
+ throws Exception {
+ long nowMs = System.currentTimeMillis();
+ Map<String, Set<Pair<TableRebalanceRetryConfig, Long>>> candidates = new HashMap<>();
+ // If the job started most recently has already completed, then skip retry for the table.
+ Pair<String, Long> latestStartedJob = null;
+ Pair<String, Long> latestCompletedJob = null;
+ // The processing order of job metadata from the given Map is not deterministic. Track the completed original
+ // jobs so that we can simply skip the retry jobs belonging to the completed original jobs.
+ Map<String, String> completedOriginalJobs = new HashMap<>();
+ for (Map.Entry<String, Map<String, String>> entry : allJobMetadata.entrySet()) {
+ String jobId = entry.getKey();
+ Map<String, String> jobMetadata = entry.getValue();
+ long statsUpdatedAt = Long.parseLong(jobMetadata.get(CommonConstants.ControllerJob.SUBMISSION_TIME_MS));
+ String jobStatsInStr = jobMetadata.get(RebalanceJobConstants.JOB_METADATA_KEY_REBALANCE_PROGRESS_STATS);
+ if (StringUtils.isEmpty(jobStatsInStr)) {
+ LOGGER.info("Skip rebalance job: {} as it has no job progress stats", jobId);
+ continue;
+ }
+ String retryCfgInStr = jobMetadata.get(RebalanceJobConstants.JOB_METADATA_KEY_REBALANCE_RETRY_CONFIG);
+ if (StringUtils.isEmpty(retryCfgInStr)) {
+ LOGGER.info("Skip rebalance job: {} as it has no job retry configs", jobId);
+ continue;
+ }
+ TableRebalanceProgressStats progress = JsonUtils.stringToObject(jobStatsInStr, TableRebalanceProgressStats.class);
+ TableRebalanceRetryConfig retryConfig = JsonUtils.stringToObject(retryCfgInStr, TableRebalanceRetryConfig.class);
+ long jobStartTime = progress.getStartTimeMs();
+ if (latestStartedJob == null || latestStartedJob.getRight() < jobStartTime) {
+ latestStartedJob = Pair.of(jobId, jobStartTime);
+ }
+ String originalJobId = retryConfig.getOriginalJobId();
+ RebalanceResult.Status jobStatus = RebalanceResult.Status.valueOf(progress.getStatus());
+ if (jobStatus == RebalanceResult.Status.DONE || jobStatus == RebalanceResult.Status.NO_OP) {
+ LOGGER.info("Skip rebalance job: {} as it has completed with status: {}", jobId, jobStatus);
+ completedOriginalJobs.put(originalJobId, jobId);
+ if (latestCompletedJob == null || latestCompletedJob.getRight() < jobStartTime) {
+ latestCompletedJob = Pair.of(jobId, jobStartTime);
+ }
+ continue;
+ }
+ if (jobStatus == RebalanceResult.Status.FAILED) {
+ LOGGER.info("Found failed rebalance job: {} for original job: {}", jobId, originalJobId);
+ candidates.computeIfAbsent(originalJobId, (k) -> new HashSet<>()).add(Pair.of(retryConfig, jobStartTime));
+ continue;
+ }
+ // Check if an IN_PROGRESS job is still actively running.
+ long heartbeatTimeoutMs = retryConfig.getConfig().getHeartbeatTimeoutInMs();
+ if (nowMs - statsUpdatedAt < heartbeatTimeoutMs) {
+ LOGGER.info("Rebalance job: {} is actively running with status updated at: {} within timeout: {}. Skip "
+ + "retry for table: {}", jobId, statsUpdatedAt, heartbeatTimeoutMs, tableNameWithType);
+ return Collections.emptyMap();
+ }
+ // The job is considered failed, but it's possible it is still running, then we might end up with more than one
+ // rebalance jobs running in parallel for a table. The rebalance algorithm is idempotent, so this should be fine
+ // for the correctness.
+ LOGGER.info("Found stuck rebalance job: {} for original job: {}", jobId, originalJobId);
+ candidates.computeIfAbsent(originalJobId, (k) -> new HashSet<>()).add(Pair.of(retryConfig, jobStartTime));
Review Comment:
yup, method tryStopExistingJobs() does this
--
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
Re: [PR] Add support to retry failed table rebalance [pinot]
Posted by "klsince (via GitHub)" <gi...@apache.org>.
klsince commented on code in PR #11740:
URL: https://github.com/apache/pinot/pull/11740#discussion_r1348091747
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManager.java:
##########
@@ -2063,28 +2066,76 @@ public boolean addNewReloadAllSegmentsJob(String tableNameWithType, String jobId
ZKMetadataProvider.constructPropertyStorePathForControllerJob(ControllerJobType.RELOAD_SEGMENT));
}
+ /**
+ * Adds a new job metadata for controller job like table rebalance or reload into ZK
+ * @param jobId job's UUID
+ * @param jobMetadata the job metadata
+ * @param jobResourcePath zkPath to add the new job metadata
+ * @return boolean representing success / failure of the ZK write step
+ */
public boolean addControllerJobToZK(String jobId, Map<String, String> jobMetadata, String jobResourcePath) {
+ return addControllerJobToZK(jobId, jobMetadata, jobResourcePath, prev -> true);
+ }
+
+ /**
+ * Adds a new job metadata for controller job like table rebalance or reload into ZK
+ * @param jobId job's UUID
+ * @param jobMetadata the job metadata
+ * @param jobResourcePath zkPath to add the new job metadata
+ * @param prevJobMetadataChecker to check the previous job metadata before adding new one
Review Comment:
like in this PR, we check if prev status is FAILED or not. If FAILED, the rebalance aborts itself.
--
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
Re: [PR] Add support to retry failed table rebalance [pinot]
Posted by "zhtaoxiang (via GitHub)" <gi...@apache.org>.
zhtaoxiang commented on code in PR #11740:
URL: https://github.com/apache/pinot/pull/11740#discussion_r1346557366
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/rebalance/TableRebalancer.java:
##########
@@ -137,6 +140,23 @@ public static String createUniqueRebalanceJobIdentifier() {
}
public RebalanceResult rebalance(TableConfig tableConfig, RebalanceConfig rebalanceConfig) {
+ long startTime = System.currentTimeMillis();
+ String tableNameWithType = tableConfig.getTableName();
+ try {
+ RebalanceResult result = doRebalance(tableConfig, rebalanceConfig);
+ if (_controllerMetrics != null && result.getStatus() == RebalanceResult.Status.FAILED) {
Review Comment:
should we also emit success metric so that it's easy to figure out whether a rebalance succeeds or fails or enters an unknown state?
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManager.java:
##########
@@ -2064,27 +2067,53 @@ public boolean addNewReloadAllSegmentsJob(String tableNameWithType, String jobId
}
public boolean addControllerJobToZK(String jobId, Map<String, String> jobMetadata, String jobResourcePath) {
+ return addControllerJobToZK(jobId, jobMetadata, jobResourcePath, prev -> true);
+ }
+
+ public boolean addControllerJobToZK(String jobId, Map<String, String> jobMetadata, String jobResourcePath,
+ Predicate<Map<String, String>> prevJobMetadataChecker) {
Preconditions.checkState(jobMetadata.get(CommonConstants.ControllerJob.SUBMISSION_TIME_MS) != null,
"Submission Time in JobMetadata record not set. Cannot expire these records");
Stat stat = new Stat();
- ZNRecord tableJobsZnRecord = _propertyStore.get(jobResourcePath, stat, AccessOption.PERSISTENT);
- if (tableJobsZnRecord != null) {
- Map<String, Map<String, String>> tasks = tableJobsZnRecord.getMapFields();
- tasks.put(jobId, jobMetadata);
- if (tasks.size() > CommonConstants.ControllerJob.MAXIMUM_CONTROLLER_JOBS_IN_ZK) {
- tasks = tasks.entrySet().stream().sorted((v1, v2) -> Long.compare(
+ ZNRecord jobsZnRecord = _propertyStore.get(jobResourcePath, stat, AccessOption.PERSISTENT);
+ if (jobsZnRecord != null) {
+ Map<String, Map<String, String>> jobMetadataMap = jobsZnRecord.getMapFields();
+ Map<String, String> prevJobMetadata = jobMetadataMap.get(jobId);
+ if (!prevJobMetadataChecker.test(prevJobMetadata)) {
+ return false;
+ }
+ jobMetadataMap.put(jobId, jobMetadata);
+ if (jobMetadataMap.size() > CommonConstants.ControllerJob.MAXIMUM_CONTROLLER_JOBS_IN_ZK) {
+ jobMetadataMap = jobMetadataMap.entrySet().stream().sorted((v1, v2) -> Long.compare(
Long.parseLong(v2.getValue().get(CommonConstants.ControllerJob.SUBMISSION_TIME_MS)),
Long.parseLong(v1.getValue().get(CommonConstants.ControllerJob.SUBMISSION_TIME_MS))))
.collect(Collectors.toList()).subList(0, CommonConstants.ControllerJob.MAXIMUM_CONTROLLER_JOBS_IN_ZK)
.stream().collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
}
- tableJobsZnRecord.setMapFields(tasks);
- return _propertyStore.set(jobResourcePath, tableJobsZnRecord, stat.getVersion(), AccessOption.PERSISTENT);
+ jobsZnRecord.setMapFields(jobMetadataMap);
+ return _propertyStore.set(jobResourcePath, jobsZnRecord, stat.getVersion(), AccessOption.PERSISTENT);
} else {
- tableJobsZnRecord = new ZNRecord(jobResourcePath);
- tableJobsZnRecord.setMapField(jobId, jobMetadata);
- return _propertyStore.set(jobResourcePath, tableJobsZnRecord, AccessOption.PERSISTENT);
+ jobsZnRecord = new ZNRecord(jobResourcePath);
+ jobsZnRecord.setMapField(jobId, jobMetadata);
+ return _propertyStore.set(jobResourcePath, jobsZnRecord, AccessOption.PERSISTENT);
+ }
+ }
+
+ public boolean updateAllJobsForTable(String tableNameWithType, String jobResourcePath,
Review Comment:
can we please add java doc to public methods?
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManager.java:
##########
@@ -2064,27 +2067,53 @@ public boolean addNewReloadAllSegmentsJob(String tableNameWithType, String jobId
}
public boolean addControllerJobToZK(String jobId, Map<String, String> jobMetadata, String jobResourcePath) {
+ return addControllerJobToZK(jobId, jobMetadata, jobResourcePath, prev -> true);
+ }
+
+ public boolean addControllerJobToZK(String jobId, Map<String, String> jobMetadata, String jobResourcePath,
Review Comment:
can we please add java doc to public methods?
##########
pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotTableRestletResource.java:
##########
@@ -615,11 +615,14 @@ public RebalanceResult rebalance(
@QueryParam("externalViewCheckIntervalInMs") long externalViewCheckIntervalInMs,
@ApiParam(value = "How long to wait till external view converges with ideal states") @DefaultValue("3600000")
@QueryParam("externalViewStabilizationTimeoutInMs") long externalViewStabilizationTimeoutInMs,
+ @ApiParam(value = "How often to make a status update (i.e. heartbeat)") @DefaultValue("300000")
Review Comment:
we also need to update the `rebalanceStatus` API to include retry information
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/rebalance/RebalanceChecker.java:
##########
@@ -0,0 +1,258 @@
+/**
+ * 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 com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Preconditions;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ExecutorService;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.commons.lang3.tuple.Pair;
+import org.apache.pinot.common.metadata.ZKMetadataProvider;
+import org.apache.pinot.common.metadata.controllerjob.ControllerJobType;
+import org.apache.pinot.common.metrics.ControllerMeter;
+import org.apache.pinot.common.metrics.ControllerMetrics;
+import org.apache.pinot.controller.ControllerConf;
+import org.apache.pinot.controller.LeadControllerManager;
+import org.apache.pinot.controller.helix.core.PinotHelixResourceManager;
+import org.apache.pinot.controller.helix.core.periodictask.ControllerPeriodicTask;
+import org.apache.pinot.spi.config.table.TableConfig;
+import org.apache.pinot.spi.utils.CommonConstants;
+import org.apache.pinot.spi.utils.JsonUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+/**
+ * Periodic task to check whether a user triggered rebalance job is completed or not, and retry if failed. The retry
+ * job is started with the same rebalance configs provided by the user and does best effort to stop the other jobs
+ * for the same table. This task can be configured to just check failures and report metrics, and not to do retry.
+ */
+public class RebalanceChecker extends ControllerPeriodicTask<Void> {
+ private static final Logger LOGGER = LoggerFactory.getLogger(RebalanceChecker.class);
+ private final boolean _checkOnly;
+ private final ExecutorService _executorService;
+
+ public RebalanceChecker(PinotHelixResourceManager pinotHelixResourceManager,
+ LeadControllerManager leadControllerManager, ControllerConf config, ControllerMetrics controllerMetrics,
+ ExecutorService executorService) {
+ super(RebalanceChecker.class.getSimpleName(), config.getRebalanceCheckerFrequencyInSeconds(),
+ config.getRebalanceCheckerInitialDelayInSeconds(), pinotHelixResourceManager, leadControllerManager,
+ controllerMetrics);
+ _checkOnly = config.isRebalanceCheckerCheckOnly();
+ _executorService = executorService;
+ }
+
+ @Override
+ protected void processTable(String tableNameWithType) {
+ _executorService.submit(() -> {
+ try {
+ TableConfig tableConfig = _pinotHelixResourceManager.getTableConfig(tableNameWithType);
+ Preconditions.checkState(tableConfig != null, "Failed to find table config for table: {}", tableNameWithType);
+ retryRebalanceTable(tableNameWithType, tableConfig);
+ } catch (Throwable t) {
+ LOGGER.error("Failed to retry rebalance for table: {}", tableNameWithType, t);
+ }
+ });
+ }
+
+ @VisibleForTesting
+ void retryRebalanceTable(String tableNameWithType, TableConfig tableConfig)
+ throws Exception {
+ // Get all rebalance jobs as tracked in ZK for this table to check if there is any failure to retry.
+ Map<String, Map<String, String>> allJobMetadata = _pinotHelixResourceManager.getAllJobsForTable(tableNameWithType,
+ Collections.singleton(ControllerJobType.TABLE_REBALANCE));
+ if (allJobMetadata.isEmpty()) {
+ LOGGER.info("No rebalance job has been triggered for table: {}. Skip retry", tableNameWithType);
+ return;
+ }
+ // Skip retry for the table if rebalance job is still running or has completed, in specific:
+ // 1) Skip retry if any rebalance job is actively running. Being actively running means the job is at IN_PROGRESS
+ // status, and has updated its status kept in ZK within the heartbeat timeout. It's possible that more than one
+ // rebalance jobs are running for the table, but that's fine with idempotent rebalance algorithm.
+ // 2) Skip retry if the most recently started rebalance job has completed with DONE or NO_OP. It's possible that
+ // jobs started earlier may be still running, but they are ignored here.
+ //
+ // Otherwise, we can get a list of failed rebalance jobs, i.e. those at FAILED status; or IN_PROGRESS status but
+ // haven't updated their status kept in ZK within the heartbeat timeout. For those candidate jobs to retry:
+ // 1) Firstly, group them by the original jobIds they retry for so that we can skip those exceeded maxRetry.
+ // 2) For the remaining jobs, we take the one started most recently and retry it with its original configs.
+ // 3) If configured, we can abort the other rebalance jobs for the table by setting their status to FAILED.
+ Map<String/*original jobId*/, Set<Pair<TableRebalanceRetryConfig/*original or retry jobId*/, Long
+ /*job startTime*/>>> candidateJobs = getCandidateJobs(tableNameWithType, allJobMetadata);
+ if (candidateJobs.isEmpty()) {
+ LOGGER.info("Found no failed rebalance jobs for table: {}. Skip retry", tableNameWithType);
+ return;
+ }
+ _controllerMetrics.addMeteredTableValue(tableNameWithType, ControllerMeter.TABLE_REBALANCE_FAILURE_DETECTED, 1L);
+ if (_checkOnly) {
+ LOGGER.info("Found failed rebalance jobs: {} for table: {}, but check only", candidateJobs.keySet(),
+ tableNameWithType);
+ return;
+ }
+ TableRebalanceRetryConfig retryConfig = getLatestJob(candidateJobs);
+ if (retryConfig == null) {
+ LOGGER.info("Rebalance is retried enough times for table: {}. Skip retry", tableNameWithType);
+ return;
+ }
+ tryStopExistingJobs(tableNameWithType, _pinotHelixResourceManager);
+ String newRetryJobId = TableRebalancer.createUniqueRebalanceJobIdentifier();
+ // Reset the JOB_ID of rebalanceConfig to be the new retry job ID.
+ RebalanceConfig rebalanceConfig = retryConfig.getConfig();
+ rebalanceConfig.setJobId(newRetryJobId);
+ ZkBasedTableRebalanceObserver observer = new ZkBasedTableRebalanceObserver(tableNameWithType, newRetryJobId,
+ TableRebalanceRetryConfig.forRetryRun(retryConfig.getOriginalJobId(), rebalanceConfig,
+ retryConfig.getRetryNum() + 1), _pinotHelixResourceManager);
+ LOGGER.info("Retry original rebalance job: {} for table: {} with new retry job: {}", retryConfig.getOriginalJobId(),
+ tableNameWithType, newRetryJobId);
+ _controllerMetrics.addMeteredTableValue(tableNameWithType, ControllerMeter.TABLE_REBALANCE_RETRY, 1L);
+ RebalanceResult result =
+ _pinotHelixResourceManager.rebalanceTable(tableNameWithType, tableConfig, newRetryJobId, rebalanceConfig,
+ observer);
+ LOGGER.info("Retry job: {} for table: {} is done with result: {}", newRetryJobId, tableNameWithType, result);
+ }
+
+ private static void tryStopExistingJobs(String tableNameWithType,
+ PinotHelixResourceManager pinotHelixResourceManager) {
+ pinotHelixResourceManager.updateAllJobsForTable(tableNameWithType,
+ ZKMetadataProvider.constructPropertyStorePathForControllerJob(ControllerJobType.TABLE_REBALANCE),
+ jobMetadata -> updateJobMetadata(tableNameWithType, jobMetadata));
+ }
+
+ private static void updateJobMetadata(String tableNameWithType, Map<String, String> jobMetadata) {
+ String jobId = jobMetadata.get(CommonConstants.ControllerJob.JOB_ID);
+ try {
+ String jobStatsInStr = jobMetadata.get(RebalanceJobConstants.JOB_METADATA_KEY_REBALANCE_PROGRESS_STATS);
+ TableRebalanceProgressStats jobStats = JsonUtils.stringToObject(jobStatsInStr, TableRebalanceProgressStats.class);
+ LOGGER.info("Set status of rebalance job: {} for table: {} from: {} to: FAILED", jobId, tableNameWithType,
+ jobStats.getStatus());
+ jobStats.setStatus(RebalanceResult.Status.FAILED.toString());
+ jobMetadata.put(RebalanceJobConstants.JOB_METADATA_KEY_REBALANCE_PROGRESS_STATS,
+ JsonUtils.objectToString(jobStats));
+ } catch (Exception e) {
+ LOGGER.error("Failed to update status of rebalance job: {} for table: {}", jobId, tableNameWithType, e);
+ }
+ }
+
+ @VisibleForTesting
+ static TableRebalanceRetryConfig getLatestJob(Map<String, Set<Pair<TableRebalanceRetryConfig, Long>>> candidateJobs) {
+ Pair<TableRebalanceRetryConfig, Long> candidateJobRun = null;
+ for (Map.Entry<String, Set<Pair<TableRebalanceRetryConfig, Long>>> entry : candidateJobs.entrySet()) {
+ // The job configs from all retry jobs are same, as the same set of job configs is used to do retry.
+ // The job metadata kept in ZK is cleaned by submission time order gradually, so we can't compare Set.size()
+ // against maxRetry, but check retryNum of each run to see if retries have exceeded limit.
+ Set<Pair<TableRebalanceRetryConfig, Long>> jobRuns = entry.getValue();
+ int maxRetry = jobRuns.iterator().next().getLeft().getConfig().getMaxRetry();
+ Pair<TableRebalanceRetryConfig, Long> latestJobRun = null;
+ for (Pair<TableRebalanceRetryConfig, Long> jobRun : jobRuns) {
+ if (jobRun.getLeft().getRetryNum() >= maxRetry) {
+ latestJobRun = null;
+ break;
+ }
+ if (latestJobRun == null || latestJobRun.getRight() < jobRun.getRight()) {
+ latestJobRun = jobRun;
+ }
+ }
+ if (latestJobRun == null) {
+ LOGGER.info("Rebalance job: {} had exceeded maxRetry: {}. Skip retry", entry.getKey(), maxRetry);
+ continue;
+ }
+ if (candidateJobRun == null || candidateJobRun.getRight() < latestJobRun.getRight()) {
+ candidateJobRun = latestJobRun;
+ }
+ }
+ return candidateJobRun == null ? null : candidateJobRun.getLeft();
+ }
+
+ @VisibleForTesting
+ static Map<String, Set<Pair<TableRebalanceRetryConfig, Long>>> getCandidateJobs(String tableNameWithType,
+ Map<String, Map<String, String>> allJobMetadata)
+ throws Exception {
+ long nowMs = System.currentTimeMillis();
+ Map<String, Set<Pair<TableRebalanceRetryConfig, Long>>> candidates = new HashMap<>();
+ // If the job started most recently has already completed, then skip retry for the table.
+ Pair<String, Long> latestStartedJob = null;
+ Pair<String, Long> latestCompletedJob = null;
+ // The processing order of job metadata from the given Map is not deterministic. Track the completed original
+ // jobs so that we can simply skip the retry jobs belonging to the completed original jobs.
+ Map<String, String> completedOriginalJobs = new HashMap<>();
+ for (Map.Entry<String, Map<String, String>> entry : allJobMetadata.entrySet()) {
+ String jobId = entry.getKey();
+ Map<String, String> jobMetadata = entry.getValue();
+ long statsUpdatedAt = Long.parseLong(jobMetadata.get(CommonConstants.ControllerJob.SUBMISSION_TIME_MS));
+ String jobStatsInStr = jobMetadata.get(RebalanceJobConstants.JOB_METADATA_KEY_REBALANCE_PROGRESS_STATS);
+ if (StringUtils.isEmpty(jobStatsInStr)) {
+ LOGGER.info("Skip rebalance job: {} as it has no job progress stats", jobId);
+ continue;
+ }
+ String retryCfgInStr = jobMetadata.get(RebalanceJobConstants.JOB_METADATA_KEY_REBALANCE_RETRY_CONFIG);
+ if (StringUtils.isEmpty(retryCfgInStr)) {
+ LOGGER.info("Skip rebalance job: {} as it has no job retry configs", jobId);
+ continue;
+ }
+ TableRebalanceProgressStats progress = JsonUtils.stringToObject(jobStatsInStr, TableRebalanceProgressStats.class);
+ TableRebalanceRetryConfig retryConfig = JsonUtils.stringToObject(retryCfgInStr, TableRebalanceRetryConfig.class);
+ long jobStartTime = progress.getStartTimeMs();
+ if (latestStartedJob == null || latestStartedJob.getRight() < jobStartTime) {
+ latestStartedJob = Pair.of(jobId, jobStartTime);
+ }
+ String originalJobId = retryConfig.getOriginalJobId();
+ RebalanceResult.Status jobStatus = RebalanceResult.Status.valueOf(progress.getStatus());
+ if (jobStatus == RebalanceResult.Status.DONE || jobStatus == RebalanceResult.Status.NO_OP) {
+ LOGGER.info("Skip rebalance job: {} as it has completed with status: {}", jobId, jobStatus);
+ completedOriginalJobs.put(originalJobId, jobId);
+ if (latestCompletedJob == null || latestCompletedJob.getRight() < jobStartTime) {
+ latestCompletedJob = Pair.of(jobId, jobStartTime);
+ }
+ continue;
+ }
+ if (jobStatus == RebalanceResult.Status.FAILED) {
+ LOGGER.info("Found failed rebalance job: {} for original job: {}", jobId, originalJobId);
+ candidates.computeIfAbsent(originalJobId, (k) -> new HashSet<>()).add(Pair.of(retryConfig, jobStartTime));
+ continue;
+ }
+ // Check if an IN_PROGRESS job is still actively running.
+ long heartbeatTimeoutMs = retryConfig.getConfig().getHeartbeatTimeoutInMs();
+ if (nowMs - statsUpdatedAt < heartbeatTimeoutMs) {
+ LOGGER.info("Rebalance job: {} is actively running with status updated at: {} within timeout: {}. Skip "
+ + "retry for table: {}", jobId, statsUpdatedAt, heartbeatTimeoutMs, tableNameWithType);
+ return Collections.emptyMap();
+ }
+ // The job is considered failed, but it's possible it is still running, then we might end up with more than one
+ // rebalance jobs running in parallel for a table. The rebalance algorithm is idempotent, so this should be fine
+ // for the correctness.
+ LOGGER.info("Found stuck rebalance job: {} for original job: {}", jobId, originalJobId);
+ candidates.computeIfAbsent(originalJobId, (k) -> new HashSet<>()).add(Pair.of(retryConfig, jobStartTime));
Review Comment:
Should we set the job status to failed if it has already timed out?
--
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
Re: [PR] Add support to retry failed table rebalance [pinot]
Posted by "klsince (via GitHub)" <gi...@apache.org>.
klsince commented on code in PR #11740:
URL: https://github.com/apache/pinot/pull/11740#discussion_r1361337217
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/rebalance/TableRebalancer.java:
##########
@@ -447,6 +467,9 @@ public RebalanceResult rebalance(TableConfig tableConfig, RebalanceConfig rebala
// Check version and update IdealState
try {
+ if (_tableRebalanceObserver.isAborted()) {
+ throw new RuntimeException(String.format("Rebalance for table: %s has been aborted", tableNameWithType));
Review Comment:
added, so that metrics can track aborted/cancelled cases separately
--
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
Re: [PR] Add support to retry failed table rebalance [pinot]
Posted by "klsince (via GitHub)" <gi...@apache.org>.
klsince commented on code in PR #11740:
URL: https://github.com/apache/pinot/pull/11740#discussion_r1361245932
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManager.java:
##########
@@ -1992,19 +1994,24 @@ public Map<String, Map<String, String>> getAllJobsForTable(String tableNameWithT
for (ControllerJobType jobType : jobTypes) {
String jobsResourcePath = ZKMetadataProvider.constructPropertyStorePathForControllerJob(jobType);
try {
+ if (!_propertyStore.exists(jobsResourcePath, -1)) {
Review Comment:
as tested, propertyStore.get() threw ZkNoNodeException. The current try-catch would continue the for-loop upon this exception, but the warn logs were kinda annoying, so I added the explicit existence check.
looks like no AccessOption is `-1`, but changed to PERSISTENT anyway
--
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
Re: [PR] Add support to retry failed table rebalance [pinot]
Posted by "klsince (via GitHub)" <gi...@apache.org>.
klsince commented on code in PR #11740:
URL: https://github.com/apache/pinot/pull/11740#discussion_r1361245932
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManager.java:
##########
@@ -1992,19 +1994,24 @@ public Map<String, Map<String, String>> getAllJobsForTable(String tableNameWithT
for (ControllerJobType jobType : jobTypes) {
String jobsResourcePath = ZKMetadataProvider.constructPropertyStorePathForControllerJob(jobType);
try {
+ if (!_propertyStore.exists(jobsResourcePath, -1)) {
Review Comment:
as tested, propertyStore.get() threw ZkNoNodeException. The current try-catch would continue the for-loop upon this exception, but the warn logs were kinda annoying, so I added the explicit existence check.
--
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
Re: [PR] Add support to retry failed table rebalance [pinot]
Posted by "klsince (via GitHub)" <gi...@apache.org>.
klsince commented on code in PR #11740:
URL: https://github.com/apache/pinot/pull/11740#discussion_r1355856100
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/rebalance/ZkBasedTableRebalanceObserver.java:
##########
@@ -126,30 +143,73 @@ public int getNumUpdatesToZk() {
return _numUpdatesToZk;
}
+ @VisibleForTesting
+ TableRebalanceRetryConfig getTableRebalanceJobRetryConfig() {
+ return _tableRebalanceJobRetryConfig;
+ }
+
private void trackStatsInZk() {
+ Map<String, String> jobMetadata =
+ createJobMetadata(_tableNameWithType, _rebalanceJobId, _tableRebalanceProgressStats,
+ _tableRebalanceJobRetryConfig);
+ _pinotHelixResourceManager.addControllerJobToZK(_rebalanceJobId, jobMetadata,
+ ZKMetadataProvider.constructPropertyStorePathForControllerJob(ControllerJobType.TABLE_REBALANCE),
+ prevJobMetadata -> {
+ // Abort the job when we're sure it has failed, otherwise continue to update the status.
+ if (prevJobMetadata == null) {
+ return true;
+ }
+ String prevStatsInStr = prevJobMetadata.get(RebalanceJobConstants.JOB_METADATA_KEY_REBALANCE_PROGRESS_STATS);
+ TableRebalanceProgressStats prevStats;
+ try {
+ prevStats = JsonUtils.stringToObject(prevStatsInStr, TableRebalanceProgressStats.class);
+ } catch (JsonProcessingException ignore) {
+ return true;
+ }
+ boolean jobFailed =
+ prevStats != null && RebalanceResult.Status.FAILED.toString().equals(prevStats.getStatus());
Review Comment:
good point!
--
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
Re: [PR] Add support to retry failed table rebalance [pinot]
Posted by "klsince (via GitHub)" <gi...@apache.org>.
klsince commented on code in PR #11740:
URL: https://github.com/apache/pinot/pull/11740#discussion_r1355832863
##########
pinot-controller/src/main/java/org/apache/pinot/controller/ControllerConf.java:
##########
@@ -199,6 +199,9 @@ public static class ControllerPeriodicTasksConf {
// RealtimeSegmentRelocator has been rebranded as SegmentRelocator
public static final String DEPRECATED_REALTIME_SEGMENT_RELOCATION_INITIAL_DELAY_IN_SECONDS =
"controller.realtimeSegmentRelocation.initialDelayInSeconds";
+ public static final String REBALANCE_CHECKER_INITIAL_DELAY_IN_SECONDS =
+ "controller.rebalanceChecker.initialDelayInSeconds";
+ public static final String REBALANCE_CHECKER_CHECK_ONLY = "controller.rebalanceChecker.checkOnly";
Review Comment:
I'm fine with removing this feature flag to be simple. I thought there might be cases where users just want to alert when this checker task detects rebalance failure, but retry the rebalance manually.
--
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
Re: [PR] Add support to retry failed table rebalance [pinot]
Posted by "Jackie-Jiang (via GitHub)" <gi...@apache.org>.
Jackie-Jiang commented on code in PR #11740:
URL: https://github.com/apache/pinot/pull/11740#discussion_r1363100963
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/rebalance/RebalanceChecker.java:
##########
@@ -0,0 +1,330 @@
+/**
+ * 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 com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Preconditions;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.Set;
+import java.util.concurrent.ExecutorService;
+import org.apache.commons.lang3.RandomUtils;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.commons.lang3.tuple.Pair;
+import org.apache.pinot.common.metadata.ZKMetadataProvider;
+import org.apache.pinot.common.metadata.controllerjob.ControllerJobType;
+import org.apache.pinot.common.metrics.ControllerGauge;
+import org.apache.pinot.common.metrics.ControllerMeter;
+import org.apache.pinot.common.metrics.ControllerMetrics;
+import org.apache.pinot.controller.ControllerConf;
+import org.apache.pinot.controller.LeadControllerManager;
+import org.apache.pinot.controller.helix.core.PinotHelixResourceManager;
+import org.apache.pinot.controller.helix.core.periodictask.ControllerPeriodicTask;
+import org.apache.pinot.spi.config.table.TableConfig;
+import org.apache.pinot.spi.utils.CommonConstants;
+import org.apache.pinot.spi.utils.JsonUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+/**
+ * Periodic task to check whether a user triggered rebalance job is completed or not, and retry if failed. The retry
+ * job is started with the same rebalance configs provided by the user and does best effort to stop the other jobs
+ * for the same table. This task can be configured to just check failures and report metrics, and not to do retry.
+ */
+public class RebalanceChecker extends ControllerPeriodicTask<Void> {
+ private static final Logger LOGGER = LoggerFactory.getLogger(RebalanceChecker.class);
+ private static final double RETRY_DELAY_SCALE_FACTOR = 2.0;
+ private final ExecutorService _executorService;
+
+ public RebalanceChecker(PinotHelixResourceManager pinotHelixResourceManager,
+ LeadControllerManager leadControllerManager, ControllerConf config, ControllerMetrics controllerMetrics,
+ ExecutorService executorService) {
+ super(RebalanceChecker.class.getSimpleName(), config.getRebalanceCheckerFrequencyInSeconds(),
+ config.getRebalanceCheckerInitialDelayInSeconds(), pinotHelixResourceManager, leadControllerManager,
+ controllerMetrics);
+ _executorService = executorService;
+ }
+
+ protected void processTables(List<String> tableNamesWithType, Properties periodicTaskProperties) {
+ int numTables = tableNamesWithType.size();
+ LOGGER.info("Processing {} tables in task: {}", numTables, _taskName);
+ int numTablesProcessed = 0;
+ for (String tableNameWithType : tableNamesWithType) {
+ if (!isStarted()) {
+ LOGGER.info("Task: {} is stopped, early terminate the task", _taskName);
+ break;
+ }
+ try {
+ processTable(tableNameWithType);
Review Comment:
`ZkCacheBaseDataAccessor` won't cache the ZNode under property store, so let's read it once and then only process the tables with pending jobs
--
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