[CLUS][MTD] += ClusterStateManager pending removal
- storage format changes: no
- interface changes: no
Add ClusterStateManager API to enable marking nodes as pending removal, so
they are not considered when creating the node groups for new datasets
Change-Id: I509a7eed727be1e339c1b75b75f778fd222cc0e2
Reviewed-on: https://asterix-gerrit.ics.uci.edu/1844
Sonar-Qube: Jenkins <jenkins@fulliautomatix.ics.uci.edu>
Integration-Tests: Jenkins <jenkins@fulliautomatix.ics.uci.edu>
Tested-by: Jenkins <jenkins@fulliautomatix.ics.uci.edu>
BAD: Jenkins <jenkins@fulliautomatix.ics.uci.edu>
Reviewed-by: Yingyi Bu <buyingyi@gmail.com>
Reviewed-by: abdullah alamoudi <bamousaa@gmail.com>
diff --git a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/translator/QueryTranslator.java b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/translator/QueryTranslator.java
index f7327b4..8a7d757 100644
--- a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/translator/QueryTranslator.java
+++ b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/translator/QueryTranslator.java
@@ -709,17 +709,19 @@
protected static String configureNodegroupForDataset(ICcApplicationContext appCtx, Map<String, String> hints,
String dataverseName, String datasetName, MetadataProvider metadataProvider) throws Exception {
- Set<String> allNodes = ClusterStateManager.INSTANCE.getParticipantNodes();
+ Set<String> allNodes = ClusterStateManager.INSTANCE.getParticipantNodes(true);
Set<String> selectedNodes = new LinkedHashSet<>();
String hintValue = hints.get(DatasetNodegroupCardinalityHint.NAME);
if (hintValue == null) {
selectedNodes.addAll(allNodes);
} else {
int nodegroupCardinality;
- boolean valid = DatasetHints.validate(appCtx, DatasetNodegroupCardinalityHint.NAME,
- hints.get(DatasetNodegroupCardinalityHint.NAME)).first;
+ final Pair<Boolean, String> validation = DatasetHints.validate(appCtx, DatasetNodegroupCardinalityHint.NAME,
+ hints.get(DatasetNodegroupCardinalityHint.NAME));
+ boolean valid = validation.first;
if (!valid) {
- throw new CompilationException("Incorrect use of hint:" + DatasetNodegroupCardinalityHint.NAME);
+ throw new CompilationException("Incorrect use of hint '" + DatasetNodegroupCardinalityHint.NAME +
+ "': " + validation.second);
} else {
nodegroupCardinality = Integer.parseInt(hints.get(DatasetNodegroupCardinalityHint.NAME));
}
diff --git a/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/dataset/hints/DatasetHints.java b/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/dataset/hints/DatasetHints.java
index 277587f..709f655 100644
--- a/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/dataset/hints/DatasetHints.java
+++ b/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/dataset/hints/DatasetHints.java
@@ -22,6 +22,7 @@
import java.util.Set;
import org.apache.asterix.common.dataflow.ICcApplicationContext;
+import org.apache.asterix.runtime.utils.ClusterStateManager;
import org.apache.hyracks.algebricks.common.utils.Pair;
/**
@@ -106,22 +107,20 @@
@Override
public Pair<Boolean, String> validateValue(ICcApplicationContext appCtx, String value) {
- boolean valid = true;
int intValue;
try {
intValue = Integer.parseInt(value);
if (intValue < 0) {
return new Pair<>(false, "Value must be >= 0");
}
- int numNodesInCluster = appCtx.getMetadataProperties().getNodeNames().size();
+ int numNodesInCluster = ClusterStateManager.INSTANCE.getParticipantNodes(true).size();
if (numNodesInCluster < intValue) {
return new Pair<>(false,
- "Value must be greater or equal to the existing number of nodes in cluster ("
+ "Value must be less than or equal to the available number of nodes in cluster ("
+ numNodesInCluster + ")");
}
} catch (NumberFormatException nfe) {
- valid = false;
- return new Pair<>(valid, "Inappropriate value");
+ return new Pair<>(false, "Inappropriate value");
}
return new Pair<>(true, null);
}
diff --git a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/utils/ClusterStateManager.java b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/utils/ClusterStateManager.java
index 9c96d32..da341af 100644
--- a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/utils/ClusterStateManager.java
+++ b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/utils/ClusterStateManager.java
@@ -65,6 +65,7 @@
private static final Logger LOGGER = Logger.getLogger(ClusterStateManager.class.getName());
public static final ClusterStateManager INSTANCE = new ClusterStateManager();
private final Map<String, Map<IOption, Object>> activeNcConfiguration = new HashMap<>();
+ private Set<String> pendingRemoval = new HashSet<>();
private final Cluster cluster;
private ClusterState state = ClusterState.UNUSABLE;
@@ -73,8 +74,8 @@
private boolean globalRecoveryCompleted = false;
- private Map<String, ClusterPartition[]> node2PartitionsMap = null;
- private SortedMap<Integer, ClusterPartition> clusterPartitions = null;
+ private Map<String, ClusterPartition[]> node2PartitionsMap;
+ private SortedMap<Integer, ClusterPartition> clusterPartitions;
private String currentMetadataNode = null;
private boolean metadataNodeActive = false;
@@ -101,6 +102,7 @@
}
failedNodes.add(nodeId);
ftStrategy.notifyNodeFailure(nodeId);
+ pendingRemoval.remove(nodeId);
}
public synchronized void addNCConfiguration(String nodeId, Map<IOption, Object> configuration)
@@ -238,6 +240,14 @@
return participantNodes;
}
+ public synchronized Set<String> getParticipantNodes(boolean excludePendingRemoval) {
+ Set<String> participantNodes = getParticipantNodes();
+ if (excludePendingRemoval) {
+ participantNodes.removeAll(pendingRemoval);
+ }
+ return participantNodes;
+ }
+
public synchronized AlgebricksAbsolutePartitionConstraint getClusterLocations() {
if (clusterPartitionConstraint == null) {
resetClusterPartitionConstraint();
@@ -387,4 +397,31 @@
}
}
}
+
+ public synchronized void removePending(String nodeId) {
+ if (LOGGER.isLoggable(Level.INFO)) {
+ LOGGER.info("Registering intention to remove node id " + nodeId);
+ }
+ if (!activeNcConfiguration.containsKey(nodeId)) {
+ LOGGER.warning("Cannot register unknown node " + nodeId + " for pending removal");
+ }
+ pendingRemoval.add(nodeId);
+ }
+
+ public synchronized boolean cancelRemovePending(String nodeId) {
+ if (LOGGER.isLoggable(Level.INFO)) {
+ LOGGER.info("Deregistering intention to remove node id " + nodeId);
+ }
+ if (!pendingRemoval.remove(nodeId)) {
+ LOGGER.warning("Cannot deregister intention to remove node id " + nodeId + " that was not registered");
+ return false;
+ } else {
+ return true;
+ }
+ }
+
+ public synchronized Set<String> getNodesPendingRemoval() {
+ return new HashSet<>(pendingRemoval);
+ }
+
}