[NO ISSUE][STO] Add force rebalance to dataset rebalance API
- user model changes: no
- storage format changes: no
- interface changes: no
Details:
- Add an option to force rebalance a dataset even if its
nodes have not changed.
- Add new force parameter in RebalanceApiServlet and default
it to true.
Change-Id: I520d6228ce11211844a9117cae0bb8654f489ee6
Reviewed-on: https://asterix-gerrit.ics.uci.edu/c/asterixdb/+/7803
Integration-Tests: Jenkins <jenkins@fulliautomatix.ics.uci.edu>
Tested-by: Jenkins <jenkins@fulliautomatix.ics.uci.edu>
Reviewed-by: Michael Blow <mblow@apache.org>
diff --git a/asterixdb/asterix-app/src/main/java/org/apache/asterix/api/http/server/RebalanceApiServlet.java b/asterixdb/asterix-app/src/main/java/org/apache/asterix/api/http/server/RebalanceApiServlet.java
index e7afd44..43c25f2 100644
--- a/asterixdb/asterix-app/src/main/java/org/apache/asterix/api/http/server/RebalanceApiServlet.java
+++ b/asterixdb/asterix-app/src/main/java/org/apache/asterix/api/http/server/RebalanceApiServlet.java
@@ -107,7 +107,11 @@
String dataverseName = request.getParameter("dataverseName");
String datasetName = request.getParameter("datasetName");
String nodes = request.getParameter("nodes");
-
+ boolean forceRebalance = true;
+ String force = request.getParameter("force");
+ if (force != null) {
+ forceRebalance = Boolean.parseBoolean(force);
+ }
// Parses and check target nodes.
if (nodes == null) {
sendResponse(response, HttpResponseStatus.BAD_REQUEST, "nodes are not given");
@@ -133,7 +137,8 @@
return;
}
// Schedules a rebalance task and wait for its completion.
- CountDownLatch terminated = scheduleRebalance(dataverseName, datasetName, targetNodes, response);
+ CountDownLatch terminated =
+ scheduleRebalance(dataverseName, datasetName, targetNodes, response, forceRebalance);
terminated.await();
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
@@ -156,10 +161,10 @@
// Schedules a rebalance task.
private synchronized CountDownLatch scheduleRebalance(String dataverseName, String datasetName,
- String[] targetNodes, IServletResponse response) {
+ String[] targetNodes, IServletResponse response, boolean force) {
CountDownLatch terminated = new CountDownLatch(1);
- Future<Void> task =
- executor.submit(() -> doRebalance(dataverseName, datasetName, targetNodes, response, terminated));
+ Future<Void> task = executor
+ .submit(() -> doRebalance(dataverseName, datasetName, targetNodes, response, terminated, force));
rebalanceTasks.add(task);
rebalanceFutureTerminated.add(terminated);
return terminated;
@@ -167,7 +172,7 @@
// Performs the actual rebalance.
private Void doRebalance(String dataverseName, String datasetName, String[] targetNodes, IServletResponse response,
- CountDownLatch terminated) {
+ CountDownLatch terminated, boolean force) {
try {
// Sets the content type.
HttpUtil.setContentType(response, HttpUtil.ContentType.APPLICATION_JSON, StandardCharsets.UTF_8);
@@ -179,11 +184,11 @@
for (Dataset dataset : datasets) {
// By the time rebalanceDataset(...) is called, the dataset could have been dropped.
// If that's the case, rebalanceDataset(...) would be a no-op.
- rebalanceDataset(dataset.getDataverseName(), dataset.getDatasetName(), targetNodes);
+ rebalanceDataset(dataset.getDataverseName(), dataset.getDatasetName(), targetNodes, force);
}
} else {
// Rebalances a given dataset from its current locations to the target nodes.
- rebalanceDataset(dataverseName, datasetName, targetNodes);
+ rebalanceDataset(dataverseName, datasetName, targetNodes, force);
}
// Sends response.
@@ -243,7 +248,8 @@
}
// Rebalances a given dataset.
- private void rebalanceDataset(String dataverseName, String datasetName, String[] targetNodes) throws Exception {
+ private void rebalanceDataset(String dataverseName, String datasetName, String[] targetNodes, boolean force)
+ throws Exception {
IHyracksClientConnection hcc = (IHyracksClientConnection) ctx.get(HYRACKS_CONNECTION_ATTR);
MetadataProvider metadataProvider = new MetadataProvider(appCtx, null);
try {
@@ -255,7 +261,7 @@
lockManager.acquireDatasetExclusiveModificationLock(metadataProvider.getLocks(),
dataverseName + '.' + datasetName);
RebalanceUtil.rebalance(dataverseName, datasetName, new LinkedHashSet<>(Arrays.asList(targetNodes)),
- metadataProvider, hcc, NoOpDatasetRebalanceCallback.INSTANCE);
+ metadataProvider, hcc, NoOpDatasetRebalanceCallback.INSTANCE, force);
} finally {
activeNotificationHandler.resume(metadataProvider);
}
diff --git a/asterixdb/asterix-app/src/main/java/org/apache/asterix/utils/RebalanceUtil.java b/asterixdb/asterix-app/src/main/java/org/apache/asterix/utils/RebalanceUtil.java
index dff0b4a..25076c2 100644
--- a/asterixdb/asterix-app/src/main/java/org/apache/asterix/utils/RebalanceUtil.java
+++ b/asterixdb/asterix-app/src/main/java/org/apache/asterix/utils/RebalanceUtil.java
@@ -93,7 +93,7 @@
*/
public static void rebalance(String dataverseName, String datasetName, Set<String> targetNcNames,
MetadataProvider metadataProvider, IHyracksClientConnection hcc,
- IDatasetRebalanceCallback datasetRebalanceCallback) throws Exception {
+ IDatasetRebalanceCallback datasetRebalanceCallback, boolean forceRebalance) throws Exception {
Dataset sourceDataset;
Dataset targetDataset;
// Executes the first Metadata transaction.
@@ -112,8 +112,7 @@
Set<String> sourceNodes = new HashSet<>(metadataProvider.findNodes(sourceDataset.getNodeGroupName()));
- // The the source nodes are identical to the target nodes.
- if (sourceNodes.equals(targetNcNames)) {
+ if (!forceRebalance && sourceNodes.equals(targetNcNames)) {
return;
}
diff --git a/asterixdb/asterix-app/src/test/java/org/apache/asterix/common/TestDataUtil.java b/asterixdb/asterix-app/src/test/java/org/apache/asterix/common/TestDataUtil.java
index d453824..f15b88f 100644
--- a/asterixdb/asterix-app/src/test/java/org/apache/asterix/common/TestDataUtil.java
+++ b/asterixdb/asterix-app/src/test/java/org/apache/asterix/common/TestDataUtil.java
@@ -167,7 +167,7 @@
lockManager.acquireDatasetExclusiveModificationLock(metadataProvider.getLocks(),
dataverseName + '.' + datasetName);
RebalanceUtil.rebalance(dataverseName, datasetName, new LinkedHashSet<>(Arrays.asList(targetNodes)),
- metadataProvider, ccAppCtx.getHcc(), NoOpDatasetRebalanceCallback.INSTANCE);
+ metadataProvider, ccAppCtx.getHcc(), NoOpDatasetRebalanceCallback.INSTANCE, false);
} finally {
activeNotificationHandler.resume(metadataProvider);
}
diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/rebalance/identical_location/identical_location.3.post.http b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/rebalance/identical_location/identical_location.3.post.http
index d21f3fc..68d4a0f 100644
--- a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/rebalance/identical_location/identical_location.3.post.http
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/rebalance/identical_location/identical_location.3.post.http
@@ -17,4 +17,4 @@
* under the License.
*/
-/admin/rebalance?dataverseName=tpch&datasetName=LineItem&nodes=%22asterix_nc2%2Casterix_nc1%22
+/admin/rebalance?dataverseName=tpch&datasetName=LineItem&nodes=%22asterix_nc2%2Casterix_nc1%22&force=false