[NO ISSUE][*DB] Explicitly create ClusterStateManager
- user model changes: no
- storage format changes: no
- interface changes: no
details:
- Previously, we create the cluster state manager using the
singleton pattern.
- After this change, cluster state manager is created
as part of the cc application start.
- To access the cluster state manager after this change,
the cc application context is used.
Change-Id: Id6532245033ac4c6f6aa9f193539944eecb832f7
Reviewed-on: https://asterix-gerrit.ics.uci.edu/1944
Tested-by: Jenkins <jenkins@fulliautomatix.ics.uci.edu>
Contrib: Jenkins <jenkins@fulliautomatix.ics.uci.edu>
Integration-Tests: Jenkins <jenkins@fulliautomatix.ics.uci.edu>
Reviewed-by: Michael Blow <mblow@apache.org>
diff --git a/asterixdb/asterix-algebra/src/main/java/org/apache/asterix/optimizer/rules/UnnestToDataScanRule.java b/asterixdb/asterix-algebra/src/main/java/org/apache/asterix/optimizer/rules/UnnestToDataScanRule.java
index 7f35d08..4550ba6 100644
--- a/asterixdb/asterix-algebra/src/main/java/org/apache/asterix/optimizer/rules/UnnestToDataScanRule.java
+++ b/asterixdb/asterix-algebra/src/main/java/org/apache/asterix/optimizer/rules/UnnestToDataScanRule.java
@@ -260,10 +260,10 @@
keyAccessExpression = null;
keyAccessScalarFunctionCallExpression = null;
}
- FeedDataSource feedDataSource = new FeedDataSource(sourceFeed, aqlId, targetDataset, feedOutputType, metaType,
- pkTypes, partitioningKeys, keyAccessScalarFunctionCallExpression, sourceFeed.getFeedId(),
- FeedRuntimeType.valueOf(subscriptionLocation), locations.split(","), context.getComputationNodeDomain(),
- feedConnection);
+ FeedDataSource feedDataSource = new FeedDataSource((MetadataProvider) context.getMetadataProvider(), sourceFeed,
+ aqlId, targetDataset, feedOutputType, metaType, pkTypes, keyAccessScalarFunctionCallExpression,
+ sourceFeed.getFeedId(), FeedRuntimeType.valueOf(subscriptionLocation), locations.split(","),
+ context.getComputationNodeDomain(), feedConnection);
feedDataSource.getProperties().put(BuiltinFeedPolicies.CONFIG_FEED_POLICY_KEY, feedPolicy);
return feedDataSource;
}
diff --git a/asterixdb/asterix-algebra/src/main/java/org/apache/asterix/translator/AbstractLangTranslator.java b/asterixdb/asterix-algebra/src/main/java/org/apache/asterix/translator/AbstractLangTranslator.java
index 1d47095..c3f01e8 100644
--- a/asterixdb/asterix-algebra/src/main/java/org/apache/asterix/translator/AbstractLangTranslator.java
+++ b/asterixdb/asterix-algebra/src/main/java/org/apache/asterix/translator/AbstractLangTranslator.java
@@ -38,7 +38,6 @@
import org.apache.asterix.metadata.dataset.hints.DatasetHints;
import org.apache.asterix.metadata.entities.Dataverse;
import org.apache.asterix.metadata.utils.MetadataConstants;
-import org.apache.asterix.runtime.utils.ClusterStateManager;
import org.apache.hyracks.algebricks.common.utils.Pair;
import org.apache.hyracks.api.exceptions.HyracksDataException;
@@ -53,7 +52,7 @@
public void validateOperation(ICcApplicationContext appCtx, Dataverse defaultDataverse, Statement stmt)
throws AsterixException {
- final IClusterStateManager clusterStateManager = ClusterStateManager.INSTANCE;
+ final IClusterStateManager clusterStateManager = appCtx.getClusterStateManager();
final IGlobalRecoveryManager globalRecoveryManager = appCtx.getGlobalRecoveryManager();
if (!(clusterStateManager.getState().equals(ClusterState.ACTIVE)
&& globalRecoveryManager.isRecoveryCompleted())) {
diff --git a/asterixdb/asterix-app/src/main/java/org/apache/asterix/api/common/AsterixHyracksIntegrationUtil.java b/asterixdb/asterix-app/src/main/java/org/apache/asterix/api/common/AsterixHyracksIntegrationUtil.java
index 2799765..71c67f4 100644
--- a/asterixdb/asterix-app/src/main/java/org/apache/asterix/api/common/AsterixHyracksIntegrationUtil.java
+++ b/asterixdb/asterix-app/src/main/java/org/apache/asterix/api/common/AsterixHyracksIntegrationUtil.java
@@ -32,10 +32,10 @@
import org.apache.asterix.common.api.IClusterManagementWork.ClusterState;
import org.apache.asterix.common.config.GlobalConfig;
import org.apache.asterix.common.config.PropertiesAccessor;
+import org.apache.asterix.common.dataflow.ICcApplicationContext;
import org.apache.asterix.common.exceptions.AsterixException;
import org.apache.asterix.hyracks.bootstrap.CCApplication;
import org.apache.asterix.hyracks.bootstrap.NCApplication;
-import org.apache.asterix.runtime.utils.ClusterStateManager;
import org.apache.commons.io.FileUtils;
import org.apache.hyracks.api.application.ICCApplication;
import org.apache.hyracks.api.application.INCApplication;
@@ -116,7 +116,7 @@
thread.join();
}
// Wait until cluster becomes active
- ClusterStateManager.INSTANCE.waitForState(ClusterState.ACTIVE);
+ ((ICcApplicationContext) cc.getApplicationContext()).getClusterStateManager().waitForState(ClusterState.ACTIVE);
hcc = new HyracksConnection(cc.getConfig().getClientListenAddress(), cc.getConfig().getClientListenPort());
this.ncs = nodeControllers.toArray(new NodeControllerService[nodeControllers.size()]);
}
diff --git a/asterixdb/asterix-app/src/main/java/org/apache/asterix/api/http/server/ClusterApiServlet.java b/asterixdb/asterix-app/src/main/java/org/apache/asterix/api/http/server/ClusterApiServlet.java
index 4faab1e..3e1a16d 100644
--- a/asterixdb/asterix-app/src/main/java/org/apache/asterix/api/http/server/ClusterApiServlet.java
+++ b/asterixdb/asterix-app/src/main/java/org/apache/asterix/api/http/server/ClusterApiServlet.java
@@ -27,8 +27,8 @@
import java.util.logging.Level;
import java.util.logging.Logger;
+import org.apache.asterix.common.dataflow.ICcApplicationContext;
import org.apache.asterix.runtime.utils.CcApplicationContext;
-import org.apache.asterix.runtime.utils.ClusterStateManager;
import org.apache.hyracks.api.config.IOption;
import org.apache.hyracks.api.config.Section;
import org.apache.hyracks.control.common.config.ConfigUtils;
@@ -58,9 +58,11 @@
protected static final String VERSION_URI_KEY = "versionUri";
protected static final String DIAGNOSTICS_URI_KEY = "diagnosticsUri";
private final ObjectMapper om = new ObjectMapper();
+ protected final ICcApplicationContext appCtx;
- public ClusterApiServlet(ConcurrentMap<String, Object> ctx, String... paths) {
+ public ClusterApiServlet(ICcApplicationContext appCtx, ConcurrentMap<String, Object> ctx, String... paths) {
super(ctx, paths);
+ this.appCtx = appCtx;
}
@Override
@@ -92,11 +94,11 @@
}
protected ObjectNode getClusterStateSummaryJSON() {
- return ClusterStateManager.INSTANCE.getClusterStateSummary();
+ return appCtx.getClusterStateManager().getClusterStateSummary();
}
protected ObjectNode getClusterStateJSON(IServletRequest request, String pathToNode) {
- ObjectNode json = ClusterStateManager.INSTANCE.getClusterStateDescription();
+ ObjectNode json = appCtx.getClusterStateManager().getClusterStateDescription();
CcApplicationContext appConfig = (CcApplicationContext) ctx.get(ASTERIX_APP_CONTEXT_INFO_ATTR);
json.putPOJO("config", ConfigUtils.getSectionOptionsForJSON(appConfig.getServiceContext().getAppConfig(),
Section.COMMON, getConfigSelector()));
diff --git a/asterixdb/asterix-app/src/main/java/org/apache/asterix/api/http/server/ClusterControllerDetailsApiServlet.java b/asterixdb/asterix-app/src/main/java/org/apache/asterix/api/http/server/ClusterControllerDetailsApiServlet.java
index 6dea30c..15f28a4 100644
--- a/asterixdb/asterix-app/src/main/java/org/apache/asterix/api/http/server/ClusterControllerDetailsApiServlet.java
+++ b/asterixdb/asterix-app/src/main/java/org/apache/asterix/api/http/server/ClusterControllerDetailsApiServlet.java
@@ -26,6 +26,7 @@
import java.util.logging.Level;
import java.util.logging.Logger;
+import org.apache.asterix.common.dataflow.ICcApplicationContext;
import org.apache.hyracks.api.client.IHyracksClientConnection;
import org.apache.hyracks.http.api.IServletRequest;
import org.apache.hyracks.http.api.IServletResponse;
@@ -41,8 +42,9 @@
private static final Logger LOGGER = Logger.getLogger(ClusterControllerDetailsApiServlet.class.getName());
private final ObjectMapper om = new ObjectMapper();
- public ClusterControllerDetailsApiServlet(ConcurrentMap<String, Object> ctx, String... paths) {
- super(ctx, paths);
+ public ClusterControllerDetailsApiServlet(ICcApplicationContext appCtx, ConcurrentMap<String, Object> ctx,
+ String... paths) {
+ super(appCtx, ctx, paths);
}
@Override
diff --git a/asterixdb/asterix-app/src/main/java/org/apache/asterix/api/http/server/DiagnosticsApiServlet.java b/asterixdb/asterix-app/src/main/java/org/apache/asterix/api/http/server/DiagnosticsApiServlet.java
index b64b6f6..a631db0 100644
--- a/asterixdb/asterix-app/src/main/java/org/apache/asterix/api/http/server/DiagnosticsApiServlet.java
+++ b/asterixdb/asterix-app/src/main/java/org/apache/asterix/api/http/server/DiagnosticsApiServlet.java
@@ -34,7 +34,8 @@
import java.util.logging.Level;
import java.util.logging.Logger;
-import org.apache.asterix.runtime.utils.ClusterStateManager;
+import org.apache.asterix.common.cluster.IClusterStateManager;
+import org.apache.asterix.common.dataflow.ICcApplicationContext;
import org.apache.hyracks.api.client.IHyracksClientConnection;
import org.apache.hyracks.http.api.IServletRequest;
import org.apache.hyracks.http.api.IServletResponse;
@@ -55,8 +56,8 @@
protected final IHyracksClientConnection hcc;
protected final ExecutorService executor;
- public DiagnosticsApiServlet(ConcurrentMap<String, Object> ctx, String[] paths) {
- super(ctx, paths);
+ public DiagnosticsApiServlet(ICcApplicationContext appCtx, ConcurrentMap<String, Object> ctx, String... paths) {
+ super(appCtx, ctx, paths);
this.om = new ObjectMapper();
this.hcc = (IHyracksClientConnection) ctx.get(HYRACKS_CONNECTION_ATTR);
this.executor = (ExecutorService) ctx.get(ServletConstants.EXECUTOR_SERVICE_ATTR);
@@ -66,7 +67,6 @@
protected void get(IServletRequest request, IServletResponse response) throws IOException {
HttpUtil.setContentType(response, HttpUtil.ContentType.APPLICATION_JSON, HttpUtil.Encoding.UTF8);
PrintWriter responseWriter = response.writer();
- ObjectNode json;
response.setStatus(HttpResponseStatus.OK);
om.enable(SerializationFeature.INDENT_OUTPUT);
try {
@@ -89,9 +89,9 @@
protected ObjectNode getClusterDiagnosticsJSON() throws Exception {
Map<String, Future<JsonNode>> ccFutureData;
ccFutureData = getCcDiagosticsFutures();
-
+ IClusterStateManager csm = appCtx.getClusterStateManager();
Map<String, Map<String, Future<JsonNode>>> ncDataMap = new HashMap<>();
- for (String nc : ClusterStateManager.INSTANCE.getParticipantNodes()) {
+ for (String nc : csm.getParticipantNodes()) {
ncDataMap.put(nc, getNcDiagnosticFutures(nc));
}
ObjectNode result = om.createObjectNode();
diff --git a/asterixdb/asterix-app/src/main/java/org/apache/asterix/api/http/server/NodeControllerDetailsApiServlet.java b/asterixdb/asterix-app/src/main/java/org/apache/asterix/api/http/server/NodeControllerDetailsApiServlet.java
index 01c59f3..ee4812a 100644
--- a/asterixdb/asterix-app/src/main/java/org/apache/asterix/api/http/server/NodeControllerDetailsApiServlet.java
+++ b/asterixdb/asterix-app/src/main/java/org/apache/asterix/api/http/server/NodeControllerDetailsApiServlet.java
@@ -29,7 +29,9 @@
import java.util.logging.Level;
import java.util.logging.Logger;
-import org.apache.asterix.runtime.utils.ClusterStateManager;
+import org.apache.asterix.common.cluster.ClusterPartition;
+import org.apache.asterix.common.cluster.IClusterStateManager;
+import org.apache.asterix.common.dataflow.ICcApplicationContext;
import org.apache.hyracks.api.client.IHyracksClientConnection;
import org.apache.hyracks.http.api.IServletRequest;
import org.apache.hyracks.http.api.IServletResponse;
@@ -48,8 +50,9 @@
private static final Logger LOGGER = Logger.getLogger(NodeControllerDetailsApiServlet.class.getName());
private final ObjectMapper om = new ObjectMapper();
- public NodeControllerDetailsApiServlet(ConcurrentMap<String, Object> ctx, String... paths) {
- super(ctx, paths);
+ public NodeControllerDetailsApiServlet(ICcApplicationContext appCtx, ConcurrentMap<String, Object> ctx,
+ String... paths) {
+ super(appCtx, ctx, paths);
om.enable(SerializationFeature.INDENT_OUTPUT);
}
@@ -205,8 +208,9 @@
String dump = hcc.getThreadDump(node);
if (dump == null) {
// check to see if this is a node that is simply down
- throw ClusterStateManager.INSTANCE.getNodePartitions(node) != null ? new IllegalStateException()
- : new IllegalArgumentException();
+ IClusterStateManager csm = appCtx.getClusterStateManager();
+ ClusterPartition[] cp = csm.getNodePartitions(node);
+ throw cp != null ? new IllegalStateException() : new IllegalArgumentException();
}
return (ObjectNode) om.readTree(dump);
}
diff --git a/asterixdb/asterix-app/src/main/java/org/apache/asterix/api/http/server/QueryServiceServlet.java b/asterixdb/asterix-app/src/main/java/org/apache/asterix/api/http/server/QueryServiceServlet.java
index fa67190..a35d191 100644
--- a/asterixdb/asterix-app/src/main/java/org/apache/asterix/api/http/server/QueryServiceServlet.java
+++ b/asterixdb/asterix-app/src/main/java/org/apache/asterix/api/http/server/QueryServiceServlet.java
@@ -39,7 +39,6 @@
import org.apache.asterix.lang.common.base.IParser;
import org.apache.asterix.lang.common.base.Statement;
import org.apache.asterix.metadata.MetadataManager;
-import org.apache.asterix.runtime.utils.ClusterStateManager;
import org.apache.asterix.translator.IStatementExecutor;
import org.apache.asterix.translator.IStatementExecutor.ResultDelivery;
import org.apache.asterix.translator.IStatementExecutor.Stats;
@@ -428,7 +427,8 @@
protected void executeStatement(String statementsText, SessionOutput sessionOutput, ResultDelivery delivery,
IStatementExecutor.Stats stats, RequestParameters param, String handleUrl, long[] outExecStartEnd)
throws Exception {
- IClusterManagementWork.ClusterState clusterState = ClusterStateManager.INSTANCE.getState();
+ IClusterManagementWork.ClusterState clusterState =
+ ((ICcApplicationContext) appCtx).getClusterStateManager().getState();
if (clusterState != IClusterManagementWork.ClusterState.ACTIVE) {
// using a plain IllegalStateException here to get into the right catch clause for a 500
throw new IllegalStateException("Cannot execute request, cluster is " + clusterState);
diff --git a/asterixdb/asterix-app/src/main/java/org/apache/asterix/api/http/server/ShutdownApiServlet.java b/asterixdb/asterix-app/src/main/java/org/apache/asterix/api/http/server/ShutdownApiServlet.java
index 38f6691..0fe6863 100644
--- a/asterixdb/asterix-app/src/main/java/org/apache/asterix/api/http/server/ShutdownApiServlet.java
+++ b/asterixdb/asterix-app/src/main/java/org/apache/asterix/api/http/server/ShutdownApiServlet.java
@@ -28,8 +28,9 @@
import java.util.logging.Level;
import java.util.logging.Logger;
+import org.apache.asterix.common.cluster.IClusterStateManager;
import org.apache.asterix.common.config.GlobalConfig;
-import org.apache.asterix.runtime.utils.ClusterStateManager;
+import org.apache.asterix.common.dataflow.ICcApplicationContext;
import org.apache.hyracks.api.client.IHyracksClientConnection;
import org.apache.hyracks.http.api.IServletRequest;
import org.apache.hyracks.http.api.IServletResponse;
@@ -49,14 +50,17 @@
public static final String NCSERVICE_PID = "ncservice_pid";
public static final String INI = "ini";
public static final String PID = "pid";
+ private final ICcApplicationContext appCtx;
- public ShutdownApiServlet(ConcurrentMap<String, Object> ctx, String[] paths) {
+ public ShutdownApiServlet(ICcApplicationContext appCtx, ConcurrentMap<String, Object> ctx, String[] paths) {
super(ctx, paths);
+ this.appCtx = appCtx;
}
@Override
protected void post(IServletRequest request, IServletResponse response) {
IHyracksClientConnection hcc = (IHyracksClientConnection) ctx.get(HYRACKS_CONNECTION_ATTR);
+ IClusterStateManager csm = appCtx.getClusterStateManager();
boolean terminateNCServices = "true".equalsIgnoreCase(request.getParameter("all"));
Thread t = new Thread(() -> {
try {
@@ -79,7 +83,7 @@
try {
jsonObject.put("status", "SHUTTING_DOWN");
jsonObject.put("date", new Date().toString());
- ObjectNode clusterState = ClusterStateManager.INSTANCE.getClusterStateDescription();
+ ObjectNode clusterState = csm.getClusterStateDescription();
ArrayNode ncs = (ArrayNode) clusterState.get("ncs");
for (int i = 0; i < ncs.size(); i++) {
ObjectNode nc = (ObjectNode) ncs.get(i);
@@ -94,7 +98,7 @@
final PrintWriter writer = response.writer();
writer.print(JSONUtil.convertNode(jsonObject));
// accept no further queries once this servlet returns
- ClusterStateManager.INSTANCE.setState(SHUTTING_DOWN);
+ csm.setState(SHUTTING_DOWN);
writer.close();
} catch (Exception e) {
GlobalConfig.ASTERIX_LOGGER.log(Level.INFO, "Exception writing response", e);
diff --git a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/message/ExecuteStatementRequestMessage.java b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/message/ExecuteStatementRequestMessage.java
index 5cee3d9..62dcede 100644
--- a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/message/ExecuteStatementRequestMessage.java
+++ b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/message/ExecuteStatementRequestMessage.java
@@ -30,6 +30,7 @@
import org.apache.asterix.api.http.server.ResultUtil;
import org.apache.asterix.app.cc.CCExtensionManager;
import org.apache.asterix.common.api.IClusterManagementWork;
+import org.apache.asterix.common.cluster.IClusterStateManager;
import org.apache.asterix.common.config.GlobalConfig;
import org.apache.asterix.common.context.IStorageComponentProvider;
import org.apache.asterix.common.dataflow.ICcApplicationContext;
@@ -41,7 +42,6 @@
import org.apache.asterix.lang.common.base.Statement;
import org.apache.asterix.messaging.CCMessageBroker;
import org.apache.asterix.metadata.MetadataManager;
-import org.apache.asterix.runtime.utils.ClusterStateManager;
import org.apache.asterix.translator.IStatementExecutor;
import org.apache.asterix.translator.IStatementExecutorContext;
import org.apache.asterix.translator.IStatementExecutorFactory;
@@ -139,7 +139,9 @@
if (ccSrv.getNodeManager().getNodeControllerState(requestNodeId) == null) {
return "Node is not registerted with the CC";
}
- final IClusterManagementWork.ClusterState clusterState = ClusterStateManager.INSTANCE.getState();
+ ICcApplicationContext appCtx = (ICcApplicationContext) ccSrv.getApplicationContext();
+ IClusterStateManager csm = appCtx.getClusterStateManager();
+ final IClusterManagementWork.ClusterState clusterState = csm.getState();
if (clusterState != IClusterManagementWork.ClusterState.ACTIVE) {
return "Cannot execute request, cluster is " + clusterState;
}
diff --git a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/nc/NCAppRuntimeContext.java b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/nc/NCAppRuntimeContext.java
index 7647881..3591bf0 100644
--- a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/nc/NCAppRuntimeContext.java
+++ b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/nc/NCAppRuntimeContext.java
@@ -131,7 +131,6 @@
private IReplicationManager replicationManager;
private IRemoteRecoveryManager remoteRecoveryManager;
private IReplicaResourcesManager replicaResourcesManager;
-
private final ILibraryManager libraryManager;
private final NCExtensionManager ncExtensionManager;
private final IStorageComponentProvider componentProvider;
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 7ed5171..365e4a3 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
@@ -56,6 +56,7 @@
import org.apache.asterix.app.result.ResultHandle;
import org.apache.asterix.app.result.ResultReader;
import org.apache.asterix.common.api.IMetadataLockManager;
+import org.apache.asterix.common.cluster.IClusterStateManager;
import org.apache.asterix.common.config.ClusterProperties;
import org.apache.asterix.common.config.DatasetConfig.DatasetType;
import org.apache.asterix.common.config.DatasetConfig.ExternalFilePendingOp;
@@ -152,7 +153,6 @@
import org.apache.asterix.om.types.ATypeTag;
import org.apache.asterix.om.types.IAType;
import org.apache.asterix.om.types.TypeSignature;
-import org.apache.asterix.runtime.utils.ClusterStateManager;
import org.apache.asterix.transaction.management.service.transaction.DatasetIdFactory;
import org.apache.asterix.translator.AbstractLangTranslator;
import org.apache.asterix.translator.CompiledStatements.CompiledDeleteStatement;
@@ -707,7 +707,8 @@
protected static String configureNodegroupForDataset(ICcApplicationContext appCtx, Map<String, String> hints,
String dataverseName, String datasetName, MetadataProvider metadataProvider) throws Exception {
- Set<String> allNodes = ClusterStateManager.INSTANCE.getParticipantNodes(true);
+ IClusterStateManager csm = appCtx.getClusterStateManager();
+ Set<String> allNodes = csm.getParticipantNodes(true);
Set<String> selectedNodes = new LinkedHashSet<>();
String hintValue = hints.get(DatasetNodegroupCardinalityHint.NAME);
if (hintValue == null) {
diff --git a/asterixdb/asterix-app/src/main/java/org/apache/asterix/hyracks/bootstrap/CCApplication.java b/asterixdb/asterix-app/src/main/java/org/apache/asterix/hyracks/bootstrap/CCApplication.java
index 6d817b8..37e0c58 100644
--- a/asterixdb/asterix-app/src/main/java/org/apache/asterix/hyracks/bootstrap/CCApplication.java
+++ b/asterixdb/asterix-app/src/main/java/org/apache/asterix/hyracks/bootstrap/CCApplication.java
@@ -54,7 +54,6 @@
import org.apache.asterix.api.http.server.VersionApiServlet;
import org.apache.asterix.app.active.ActiveNotificationHandler;
import org.apache.asterix.app.cc.CCExtensionManager;
-import org.apache.asterix.app.cc.ResourceIdManager;
import org.apache.asterix.app.external.ExternalLibraryUtils;
import org.apache.asterix.app.replication.FaultToleranceStrategyFactory;
import org.apache.asterix.common.api.AsterixThreadFactory;
@@ -79,7 +78,6 @@
import org.apache.asterix.metadata.lock.MetadataLockManager;
import org.apache.asterix.runtime.job.resource.JobCapacityController;
import org.apache.asterix.runtime.utils.CcApplicationContext;
-import org.apache.asterix.runtime.utils.ClusterStateManager;
import org.apache.asterix.translator.IStatementExecutorContext;
import org.apache.asterix.translator.IStatementExecutorFactory;
import org.apache.hyracks.api.application.ICCServiceContext;
@@ -130,7 +128,7 @@
int port = ccServiceCtx.getCCContext().getClusterControllerInfo().getClientNetPort();
hcc = new HyracksConnection(strIP, port);
ILibraryManager libraryManager = new ExternalLibraryManager();
- ResourceIdManager resourceIdManager = new ResourceIdManager();
+
IReplicationStrategy repStrategy = ClusterProperties.INSTANCE.getReplicationStrategy();
IFaultToleranceStrategy ftStrategy = FaultToleranceStrategyFactory
.create(ClusterProperties.INSTANCE.getCluster(), repStrategy, ccServiceCtx);
@@ -138,10 +136,9 @@
componentProvider = new StorageComponentProvider();
GlobalRecoveryManager globalRecoveryManager = createGlobalRecoveryManager();
statementExecutorCtx = new StatementExecutorContext();
- appCtx = new CcApplicationContext(ccServiceCtx, getHcc(), libraryManager, resourceIdManager,
- () -> MetadataManager.INSTANCE, globalRecoveryManager, ftStrategy, new ActiveNotificationHandler(),
- componentProvider, new MetadataLockManager());
- ClusterStateManager.INSTANCE.setCcAppCtx(appCtx);
+ appCtx = new CcApplicationContext(ccServiceCtx, getHcc(), libraryManager, () -> MetadataManager.INSTANCE,
+ globalRecoveryManager, ftStrategy, new ActiveNotificationHandler(), componentProvider,
+ new MetadataLockManager());
ccExtensionManager = new CCExtensionManager(getExtensions());
appCtx.setExtensionManager(ccExtensionManager);
final CCConfig ccConfig = controllerService.getCCConfig();
@@ -191,7 +188,7 @@
if (LOGGER.isLoggable(Level.INFO)) {
LOGGER.info("Stopping Asterix cluster controller");
}
- ClusterStateManager.INSTANCE.setState(SHUTTING_DOWN);
+ appCtx.getClusterStateManager().setState(SHUTTING_DOWN);
if (appCtx != null) {
((ActiveNotificationHandler) appCtx.getActiveNotificationHandler()).stop();
}
@@ -305,17 +302,17 @@
case Servlets.REBALANCE:
return new RebalanceApiServlet(ctx, paths, appCtx);
case Servlets.SHUTDOWN:
- return new ShutdownApiServlet(ctx, paths);
+ return new ShutdownApiServlet(appCtx, ctx, paths);
case Servlets.VERSION:
return new VersionApiServlet(ctx, paths);
case Servlets.CLUSTER_STATE:
- return new ClusterApiServlet(ctx, paths);
+ return new ClusterApiServlet(appCtx, ctx, paths);
case Servlets.CLUSTER_STATE_NODE_DETAIL:
- return new NodeControllerDetailsApiServlet(ctx, paths);
+ return new NodeControllerDetailsApiServlet(appCtx, ctx, paths);
case Servlets.CLUSTER_STATE_CC_DETAIL:
- return new ClusterControllerDetailsApiServlet(ctx, paths);
+ return new ClusterControllerDetailsApiServlet(appCtx, ctx, paths);
case Servlets.DIAGNOSTICS:
- return new DiagnosticsApiServlet(ctx, paths);
+ return new DiagnosticsApiServlet(appCtx, ctx, paths);
case Servlets.ACTIVE_STATS:
return new ActiveStatsApiServlet(ctx, paths, appCtx);
default:
@@ -334,7 +331,7 @@
@Override
public void startupCompleted() throws Exception {
ccServiceCtx.getControllerService().getExecutor().submit(() -> {
- ClusterStateManager.INSTANCE.waitForState(ClusterState.ACTIVE);
+ appCtx.getClusterStateManager().waitForState(ClusterState.ACTIVE);
ClusterManagerProvider.getClusterManager().notifyStartupCompleted();
return null;
});
diff --git a/asterixdb/asterix-app/src/main/java/org/apache/asterix/hyracks/bootstrap/ClusterLifecycleListener.java b/asterixdb/asterix-app/src/main/java/org/apache/asterix/hyracks/bootstrap/ClusterLifecycleListener.java
index 66f76c5..0583508 100644
--- a/asterixdb/asterix-app/src/main/java/org/apache/asterix/hyracks/bootstrap/ClusterLifecycleListener.java
+++ b/asterixdb/asterix-app/src/main/java/org/apache/asterix/hyracks/bootstrap/ClusterLifecycleListener.java
@@ -32,6 +32,7 @@
import org.apache.asterix.common.api.IClusterManagementWork;
import org.apache.asterix.common.api.IClusterManagementWorkResponse;
import org.apache.asterix.common.api.IClusterManagementWorkResponse.Status;
+import org.apache.asterix.common.cluster.IClusterStateManager;
import org.apache.asterix.common.config.ClusterProperties;
import org.apache.asterix.common.exceptions.AsterixException;
import org.apache.asterix.event.schema.cluster.Node;
@@ -42,7 +43,6 @@
import org.apache.asterix.metadata.cluster.RemoveNodeWork;
import org.apache.asterix.metadata.cluster.RemoveNodeWorkResponse;
import org.apache.asterix.runtime.utils.CcApplicationContext;
-import org.apache.asterix.runtime.utils.ClusterStateManager;
import org.apache.hyracks.api.application.IClusterLifecycleListener;
import org.apache.hyracks.api.config.IOption;
import org.apache.hyracks.api.exceptions.HyracksException;
@@ -70,10 +70,11 @@
if (LOGGER.isLoggable(Level.INFO)) {
LOGGER.info("NC: " + nodeId + " joined");
}
- ClusterStateManager.INSTANCE.addNCConfiguration(nodeId, ncConfiguration);
+ IClusterStateManager csm = appCtx.getClusterStateManager();
+ csm.addNCConfiguration(nodeId, ncConfiguration);
//if metadata node rejoining, we need to rebind the proxy connection when it is active again.
- if (!ClusterStateManager.INSTANCE.isMetadataNodeActive()) {
+ if (!csm.isMetadataNodeActive()) {
MetadataManager.INSTANCE.rebindMetadataNode();
}
@@ -99,10 +100,11 @@
if (LOGGER.isLoggable(Level.INFO)) {
LOGGER.info("NC: " + deadNode + " left");
}
- ClusterStateManager.INSTANCE.removeNCConfiguration(deadNode);
+ IClusterStateManager csm = appCtx.getClusterStateManager();
+ csm.removeNCConfiguration(deadNode);
//if metadata node failed, we need to rebind the proxy connection when it is active again
- if (!ClusterStateManager.INSTANCE.isMetadataNodeActive()) {
+ if (!csm.isMetadataNodeActive()) {
MetadataManager.INSTANCE.rebindMetadataNode();
}
}
@@ -171,8 +173,9 @@
List<String> addedNodes = new ArrayList<>();
String asterixInstanceName = ClusterProperties.INSTANCE.getCluster().getInstanceName();
+ IClusterStateManager csm = appCtx.getClusterStateManager();
for (int i = 0; i < nodesToAdd; i++) {
- Node node = ClusterStateManager.INSTANCE.getAvailableSubstitutionNode();
+ Node node = csm.getAvailableSubstitutionNode();
if (node != null) {
try {
ClusterManagerProvider.getClusterManager().addNode(appCtx, node);
diff --git a/asterixdb/asterix-app/src/main/java/org/apache/asterix/hyracks/bootstrap/ClusterWorkExecutor.java b/asterixdb/asterix-app/src/main/java/org/apache/asterix/hyracks/bootstrap/ClusterWorkExecutor.java
index 46968b4..2977a58 100644
--- a/asterixdb/asterix-app/src/main/java/org/apache/asterix/hyracks/bootstrap/ClusterWorkExecutor.java
+++ b/asterixdb/asterix-app/src/main/java/org/apache/asterix/hyracks/bootstrap/ClusterWorkExecutor.java
@@ -25,13 +25,13 @@
import java.util.logging.Logger;
import org.apache.asterix.common.api.IClusterManagementWork;
+import org.apache.asterix.common.cluster.IClusterStateManager;
import org.apache.asterix.common.dataflow.ICcApplicationContext;
import org.apache.asterix.common.exceptions.AsterixException;
import org.apache.asterix.event.schema.cluster.Node;
import org.apache.asterix.metadata.cluster.AddNodeWork;
import org.apache.asterix.metadata.cluster.ClusterManagerProvider;
import org.apache.asterix.metadata.cluster.RemoveNodeWork;
-import org.apache.asterix.runtime.utils.ClusterStateManager;
public class ClusterWorkExecutor implements Runnable {
@@ -69,9 +69,10 @@
}
}
+ IClusterStateManager csm = appCtx.getClusterStateManager();
Set<Node> addedNodes = new HashSet<>();
for (int i = 0; i < nodesToAdd; i++) {
- Node node = ClusterStateManager.INSTANCE.getAvailableSubstitutionNode();
+ Node node = csm.getAvailableSubstitutionNode();
if (node != null) {
try {
ClusterManagerProvider.getClusterManager().addNode(appCtx, node);
diff --git a/asterixdb/asterix-app/src/main/java/org/apache/asterix/utils/FeedOperations.java b/asterixdb/asterix-app/src/main/java/org/apache/asterix/utils/FeedOperations.java
index 9fc9940..cc95770 100644
--- a/asterixdb/asterix-app/src/main/java/org/apache/asterix/utils/FeedOperations.java
+++ b/asterixdb/asterix-app/src/main/java/org/apache/asterix/utils/FeedOperations.java
@@ -33,6 +33,7 @@
import org.apache.asterix.active.message.ActiveManagerMessage;
import org.apache.asterix.active.message.ActiveManagerMessage.Kind;
import org.apache.asterix.app.translator.DefaultStatementExecutorFactory;
+import org.apache.asterix.common.cluster.IClusterStateManager;
import org.apache.asterix.common.context.IStorageComponentProvider;
import org.apache.asterix.common.dataflow.ICcApplicationContext;
import org.apache.asterix.common.dataflow.LSMTreeInsertDeleteOperatorDescriptor;
@@ -66,7 +67,6 @@
import org.apache.asterix.metadata.feeds.LocationConstraint;
import org.apache.asterix.runtime.job.listener.JobEventListenerFactory;
import org.apache.asterix.runtime.job.listener.MultiTransactionJobletEventListenerFactory;
-import org.apache.asterix.runtime.utils.ClusterStateManager;
import org.apache.asterix.runtime.utils.RuntimeUtils;
import org.apache.asterix.translator.CompiledStatements;
import org.apache.asterix.translator.IStatementExecutor;
@@ -134,8 +134,10 @@
public static JobSpecification buildRemoveFeedStorageJob(MetadataProvider metadataProvider, Feed feed)
throws AsterixException {
- JobSpecification spec = RuntimeUtils.createJobSpecification(metadataProvider.getApplicationContext());
- AlgebricksAbsolutePartitionConstraint allCluster = ClusterStateManager.INSTANCE.getClusterLocations();
+ ICcApplicationContext appCtx = metadataProvider.getApplicationContext();
+ JobSpecification spec = RuntimeUtils.createJobSpecification(appCtx);
+ IClusterStateManager csm = appCtx.getClusterStateManager();
+ AlgebricksAbsolutePartitionConstraint allCluster = csm.getClusterLocations();
Set<String> nodes = new TreeSet<>();
for (String node : allCluster.getLocations()) {
nodes.add(node);
@@ -143,7 +145,7 @@
AlgebricksAbsolutePartitionConstraint locations =
new AlgebricksAbsolutePartitionConstraint(nodes.toArray(new String[nodes.size()]));
FileSplit[] feedLogFileSplits =
- FeedUtils.splitsForAdapter(feed.getDataverseName(), feed.getFeedName(), locations);
+ FeedUtils.splitsForAdapter(appCtx, feed.getDataverseName(), feed.getFeedName(), locations);
org.apache.hyracks.algebricks.common.utils.Pair<IFileSplitProvider, AlgebricksPartitionConstraint> spC =
StoragePathUtil.splitProviderAndPartitionConstraints(feedLogFileSplits);
FileRemoveOperatorDescriptor frod = new FileRemoveOperatorDescriptor(spec, spC.first, true);
@@ -273,9 +275,8 @@
}
// make connections between operators
- for (Entry<ConnectorDescriptorId,
- Pair<Pair<IOperatorDescriptor, Integer>, Pair<IOperatorDescriptor, Integer>>> entry
- : subJob.getConnectorOperatorMap().entrySet()) {
+ for (Entry<ConnectorDescriptorId, Pair<Pair<IOperatorDescriptor, Integer>,
+ Pair<IOperatorDescriptor, Integer>>> entry : subJob.getConnectorOperatorMap().entrySet()) {
ConnectorDescriptorId newId = connectorIdMapping.get(entry.getKey());
IConnectorDescriptor connDesc = jobSpec.getConnectorMap().get(newId);
Pair<IOperatorDescriptor, Integer> leftOp = entry.getValue().getLeft();
diff --git a/asterixdb/asterix-app/src/test/java/org/apache/asterix/app/bootstrap/TestNodeController.java b/asterixdb/asterix-app/src/test/java/org/apache/asterix/app/bootstrap/TestNodeController.java
index c1421c5..53a4f23 100644
--- a/asterixdb/asterix-app/src/test/java/org/apache/asterix/app/bootstrap/TestNodeController.java
+++ b/asterixdb/asterix-app/src/test/java/org/apache/asterix/app/bootstrap/TestNodeController.java
@@ -405,7 +405,10 @@
IndexType.BTREE, keyFieldNames, keyFieldSourceIndicators, keyFieldTypes, false, false, true,
MetadataUtil.PENDING_NO_OP);
List<String> nodes = Collections.singletonList(ExecutionTestUtil.integrationUtil.ncs[0].getId());
- FileSplit[] splits = SplitsAndConstraintsUtil.getIndexSplits(dataset, index.getIndexName(), nodes);
+ FileSplit[] splits = SplitsAndConstraintsUtil.getIndexSplits(
+ ((ICcApplicationContext) ExecutionTestUtil.integrationUtil.cc.getApplicationContext())
+ .getClusterStateManager(),
+ dataset, index.getIndexName(), nodes);
fileSplitProvider = new ConstantFileSplitProvider(Arrays.copyOfRange(splits, 0, 1));
}
diff --git a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/api/IApplicationContext.java b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/api/IApplicationContext.java
index 0aea84d..e1840d3 100644
--- a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/api/IApplicationContext.java
+++ b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/api/IApplicationContext.java
@@ -59,5 +59,4 @@
public ILibraryManager getLibraryManager();
IServiceContext getServiceContext();
-
}
diff --git a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/cluster/IClusterStateManager.java b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/cluster/IClusterStateManager.java
index 30675cd..b368c3b 100644
--- a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/cluster/IClusterStateManager.java
+++ b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/cluster/IClusterStateManager.java
@@ -19,12 +19,19 @@
package org.apache.asterix.common.cluster;
import java.util.Map;
+import java.util.Set;
import java.util.concurrent.TimeUnit;
import org.apache.asterix.common.api.IClusterManagementWork.ClusterState;
+import org.apache.asterix.common.dataflow.ICcApplicationContext;
import org.apache.asterix.common.exceptions.AsterixException;
+import org.apache.asterix.event.schema.cluster.Node;
+import org.apache.hyracks.algebricks.common.constraints.AlgebricksAbsolutePartitionConstraint;
import org.apache.hyracks.api.config.IOption;
import org.apache.hyracks.api.exceptions.HyracksDataException;
+import org.apache.hyracks.api.exceptions.HyracksException;
+
+import com.fasterxml.jackson.databind.node.ObjectNode;
public interface IClusterStateManager {
@@ -49,6 +56,7 @@
/**
* Updates all partitions of {@code nodeId} based on the {@code active} flag.
+ *
* @param nodeId
* @param active
* @throws HyracksDataException
@@ -93,6 +101,7 @@
/**
* Blocks until the cluster state becomes {@code state}, or timeout is exhausted.
+ *
* @return true if the desired state was reached before timeout occurred
*/
boolean waitForState(ClusterState waitForState, long timeout, TimeUnit unit)
@@ -116,4 +125,106 @@
* @throws HyracksDataException
*/
void deregisterNodePartitions(String nodeId) throws HyracksDataException;
+
+ /**
+ * @return true if cluster is active, false otherwise
+ */
+ boolean isClusterActive();
+
+ /**
+ * @return the set of participant nodes
+ */
+ Set<String> getParticipantNodes();
+
+ /**
+ * Returns the IO devices configured for a Node Controller
+ *
+ * @param nodeId
+ * unique identifier of the Node Controller
+ * @return a list of IO devices.
+ */
+ String[] getIODevices(String nodeId);
+
+ /**
+ * @return the constraint representing all the partitions of the cluster
+ */
+ AlgebricksAbsolutePartitionConstraint getClusterLocations();
+
+ /**
+ * @param excludePendingRemoval
+ * true, if the desired set shouldn't have pending removal nodes
+ * @return the set of participant nodes
+ */
+ Set<String> getParticipantNodes(boolean excludePendingRemoval);
+
+ /**
+ * @param node
+ * the node id
+ * @return the number of partitions on that node
+ */
+ int getNodePartitionsCount(String node);
+
+ /**
+ * @return a json object representing the cluster state summary
+ */
+ ObjectNode getClusterStateSummary();
+
+ /**
+ * @return a json object representing the cluster state description
+ */
+ ObjectNode getClusterStateDescription();
+
+ /**
+ * Set the cc application context
+ *
+ * @param appCtx
+ */
+ void setCcAppCtx(ICcApplicationContext appCtx);
+
+ /**
+ * @return the number of cluster nodes
+ */
+ int getNumberOfNodes();
+
+ /**
+ * Add node configuration
+ *
+ * @param nodeId
+ * @param ncConfiguration
+ * @throws HyracksException
+ */
+ void addNCConfiguration(String nodeId, Map<IOption, Object> ncConfiguration) throws HyracksException;
+
+ /**
+ * @return true if metadata node is active, false otherwise
+ */
+ boolean isMetadataNodeActive();
+
+ /**
+ * Remove configuration of a dead node
+ *
+ * @param deadNode
+ * @throws HyracksException
+ */
+ void removeNCConfiguration(String deadNode) throws HyracksException;
+
+ /**
+ * @return a substitution node or null
+ */
+ Node getAvailableSubstitutionNode();
+
+ /**
+ * Add node to the list of nodes pending removal
+ *
+ * @param nodeId
+ */
+ void removePending(String nodeId);
+
+ /**
+ * Deregister intention to remove node id
+ *
+ * @param nodeId
+ * @return
+ */
+ boolean cancelRemovePending(String nodeId);
}
diff --git a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/dataflow/ICcApplicationContext.java b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/dataflow/ICcApplicationContext.java
index 3eff214..20f685a 100644
--- a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/dataflow/ICcApplicationContext.java
+++ b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/dataflow/ICcApplicationContext.java
@@ -23,6 +23,7 @@
import org.apache.asterix.common.cluster.IClusterStateManager;
import org.apache.asterix.common.cluster.IGlobalRecoveryManager;
import org.apache.asterix.common.context.IStorageComponentProvider;
+import org.apache.asterix.common.metadata.IMetadataBootstrap;
import org.apache.asterix.common.replication.IFaultToleranceStrategy;
import org.apache.asterix.common.transactions.IResourceIdManager;
import org.apache.hyracks.api.application.ICCServiceContext;
@@ -99,6 +100,11 @@
IMetadataLockManager getMetadataLockManager();
/**
+ * @return the metadata bootstrap
+ */
+ IMetadataBootstrap getMetadataBootstrap();
+
+ /**
* @return the cluster state manager
*/
IClusterStateManager getClusterStateManager();
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/adapter/factory/GenericAdapterFactory.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/adapter/factory/GenericAdapterFactory.java
index 2eb81d4..0a47788 100644
--- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/adapter/factory/GenericAdapterFactory.java
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/adapter/factory/GenericAdapterFactory.java
@@ -25,6 +25,7 @@
import org.apache.asterix.common.api.IApplicationContext;
import org.apache.asterix.common.api.INcApplicationContext;
+import org.apache.asterix.common.dataflow.ICcApplicationContext;
import org.apache.asterix.common.library.ILibraryManager;
import org.apache.asterix.external.api.IAdapterFactory;
import org.apache.asterix.external.api.IDataFlowController;
@@ -92,8 +93,7 @@
public synchronized IDataSourceAdapter createAdapter(IHyracksTaskContext ctx, int partition)
throws HyracksDataException {
INCServiceContext serviceCtx = ctx.getJobletContext().getServiceContext();
- INcApplicationContext appCtx =
- (INcApplicationContext) serviceCtx.getApplicationContext();
+ INcApplicationContext appCtx = (INcApplicationContext) serviceCtx.getApplicationContext();
try {
restoreExternalObjects(serviceCtx, appCtx.getLibraryManager());
} catch (Exception e) {
@@ -152,15 +152,16 @@
dataParserFactory.setMetaType(metaType);
dataParserFactory.configure(configuration);
ExternalDataCompatibilityUtils.validateCompatibility(dataSourceFactory, dataParserFactory);
- configureFeedLogManager();
+ configureFeedLogManager(appCtx);
nullifyExternalObjects();
}
- private void configureFeedLogManager() throws HyracksDataException, AlgebricksException {
+ private void configureFeedLogManager(IApplicationContext appCtx) throws HyracksDataException, AlgebricksException {
this.isFeed = ExternalDataUtils.isFeed(configuration);
if (isFeed) {
- feedLogFileSplits = FeedUtils.splitsForAdapter(ExternalDataUtils.getDataverse(configuration),
- ExternalDataUtils.getFeedName(configuration), dataSourceFactory.getPartitionConstraint());
+ feedLogFileSplits = FeedUtils.splitsForAdapter((ICcApplicationContext) appCtx,
+ ExternalDataUtils.getDataverse(configuration), ExternalDataUtils.getFeedName(configuration),
+ dataSourceFactory.getPartitionConstraint());
}
}
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/api/IExternalDataSourceFactory.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/api/IExternalDataSourceFactory.java
index edda448..4c3b7e6 100644
--- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/api/IExternalDataSourceFactory.java
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/api/IExternalDataSourceFactory.java
@@ -24,9 +24,9 @@
import java.util.Map;
import java.util.Set;
-import org.apache.asterix.common.api.IApplicationContext;
+import org.apache.asterix.common.cluster.IClusterStateManager;
+import org.apache.asterix.common.dataflow.ICcApplicationContext;
import org.apache.asterix.common.exceptions.AsterixException;
-import org.apache.asterix.runtime.utils.ClusterStateManager;
import org.apache.hyracks.algebricks.common.constraints.AlgebricksAbsolutePartitionConstraint;
import org.apache.hyracks.algebricks.common.exceptions.AlgebricksException;
import org.apache.hyracks.api.application.IServiceContext;
@@ -84,9 +84,10 @@
* @return
* @throws AlgebricksException
*/
- public static AlgebricksAbsolutePartitionConstraint getPartitionConstraints(IApplicationContext appCtx,
+ public static AlgebricksAbsolutePartitionConstraint getPartitionConstraints(ICcApplicationContext appCtx,
AlgebricksAbsolutePartitionConstraint constraints, int count) throws AlgebricksException {
if (constraints == null) {
+ IClusterStateManager clusterStateManager = appCtx.getClusterStateManager();
ArrayList<String> locs = new ArrayList<>();
Set<String> stores = appCtx.getMetadataProperties().getStores().keySet();
if (stores.isEmpty()) {
@@ -97,7 +98,7 @@
Iterator<String> storeIt = stores.iterator();
while (storeIt.hasNext()) {
String node = storeIt.next();
- int numIODevices = ClusterStateManager.INSTANCE.getIODevices(node).length;
+ int numIODevices = clusterStateManager.getIODevices(node).length;
for (int k = 0; k < numIODevices; k++) {
locs.add(node);
i++;
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/rss/RSSRecordReaderFactory.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/rss/RSSRecordReaderFactory.java
index af4be91..6291ba1 100644
--- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/rss/RSSRecordReaderFactory.java
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/rss/RSSRecordReaderFactory.java
@@ -25,7 +25,7 @@
import java.util.List;
import java.util.Map;
-import org.apache.asterix.common.api.IApplicationContext;
+import org.apache.asterix.common.dataflow.ICcApplicationContext;
import org.apache.asterix.external.api.IExternalDataSourceFactory;
import org.apache.asterix.external.api.IRecordReader;
import org.apache.asterix.external.api.IRecordReaderFactory;
@@ -55,7 +55,7 @@
public AlgebricksAbsolutePartitionConstraint getPartitionConstraint() throws AlgebricksException {
int count = urls.size();
clusterLocations = IExternalDataSourceFactory.getPartitionConstraints(
- (IApplicationContext) serviceContext.getApplicationContext(), clusterLocations, count);
+ (ICcApplicationContext) serviceContext.getApplicationContext(), clusterLocations, count);
return clusterLocations;
}
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/twitter/TwitterRecordReaderFactory.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/twitter/TwitterRecordReaderFactory.java
index a0b53fd..3fb1d5f 100644
--- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/twitter/TwitterRecordReaderFactory.java
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/twitter/TwitterRecordReaderFactory.java
@@ -25,7 +25,7 @@
import java.util.logging.Level;
import java.util.logging.Logger;
-import org.apache.asterix.common.api.IApplicationContext;
+import org.apache.asterix.common.dataflow.ICcApplicationContext;
import org.apache.asterix.common.exceptions.AsterixException;
import org.apache.asterix.common.exceptions.ErrorCode;
import org.apache.asterix.external.api.IExternalDataSourceFactory;
@@ -55,20 +55,16 @@
private transient AlgebricksAbsolutePartitionConstraint clusterLocations;
private transient IServiceContext serviceCtx;
- private static final List<String> recordReaderNames = Collections.unmodifiableList(Arrays.asList(
- ExternalDataConstants.READER_TWITTER_PULL,
- ExternalDataConstants.READER_TWITTER_PUSH,
- ExternalDataConstants.READER_PUSH_TWITTER,
- ExternalDataConstants.READER_PULL_TWITTER,
- ExternalDataConstants.READER_USER_STREAM_TWITTER));
-
+ private static final List<String> recordReaderNames =
+ Collections.unmodifiableList(Arrays.asList(ExternalDataConstants.READER_TWITTER_PULL,
+ ExternalDataConstants.READER_TWITTER_PUSH, ExternalDataConstants.READER_PUSH_TWITTER,
+ ExternalDataConstants.READER_PULL_TWITTER, ExternalDataConstants.READER_USER_STREAM_TWITTER));
@Override
public DataSourceType getDataSourceType() {
return DataSourceType.RECORDS;
}
-
@Override
public List<String> getRecordReaderNames() {
return recordReaderNames;
@@ -77,8 +73,7 @@
@Override
public AlgebricksAbsolutePartitionConstraint getPartitionConstraint() throws AlgebricksException {
clusterLocations = IExternalDataSourceFactory.getPartitionConstraints(
- (IApplicationContext) serviceCtx.getApplicationContext(),
- clusterLocations, INTAKE_CARDINALITY);
+ (ICcApplicationContext) serviceCtx.getApplicationContext(), clusterLocations, INTAKE_CARDINALITY);
return clusterLocations;
}
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/stream/factory/SocketClientInputStreamFactory.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/stream/factory/SocketClientInputStreamFactory.java
index caeaa07..bfc2cb7 100644
--- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/stream/factory/SocketClientInputStreamFactory.java
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/stream/factory/SocketClientInputStreamFactory.java
@@ -25,7 +25,7 @@
import java.util.List;
import java.util.Map;
-import org.apache.asterix.common.api.IApplicationContext;
+import org.apache.asterix.common.dataflow.ICcApplicationContext;
import org.apache.asterix.common.exceptions.AsterixException;
import org.apache.asterix.external.api.AsterixInputStream;
import org.apache.asterix.external.api.IExternalDataSourceFactory;
@@ -50,7 +50,7 @@
@Override
public AlgebricksAbsolutePartitionConstraint getPartitionConstraint() throws AlgebricksException {
clusterLocations = IExternalDataSourceFactory.getPartitionConstraints(
- (IApplicationContext) serviceCtx.getApplicationContext(), clusterLocations, sockets.size());
+ (ICcApplicationContext) serviceCtx.getApplicationContext(), clusterLocations, sockets.size());
return clusterLocations;
}
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/stream/factory/TwitterFirehoseStreamFactory.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/stream/factory/TwitterFirehoseStreamFactory.java
index 12be449..c7b8633 100644
--- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/stream/factory/TwitterFirehoseStreamFactory.java
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/stream/factory/TwitterFirehoseStreamFactory.java
@@ -23,10 +23,10 @@
import java.util.List;
import java.util.Map;
+import org.apache.asterix.common.dataflow.ICcApplicationContext;
import org.apache.asterix.external.api.AsterixInputStream;
import org.apache.asterix.external.api.IInputStreamFactory;
import org.apache.asterix.external.input.stream.TwitterFirehoseInputStream;
-import org.apache.asterix.runtime.utils.ClusterStateManager;
import org.apache.hyracks.algebricks.common.constraints.AlgebricksAbsolutePartitionConstraint;
import org.apache.hyracks.api.application.IServiceContext;
import org.apache.hyracks.api.context.IHyracksTaskContext;
@@ -54,6 +54,7 @@
private static final String KEY_INGESTION_LOCATIONS = "ingestion-location";
private Map<String, String> configuration;
+ private transient IServiceContext serviceCtx;
@Override
public AlgebricksAbsolutePartitionConstraint getPartitionConstraint() {
@@ -67,10 +68,10 @@
if (ingestionCardinalityParam != null) {
count = Integer.parseInt(ingestionCardinalityParam);
}
-
+ ICcApplicationContext appCtx = (ICcApplicationContext) serviceCtx.getApplicationContext();
List<String> chosenLocations = new ArrayList<>();
String[] availableLocations = locations != null ? locations
- : ClusterStateManager.INSTANCE.getParticipantNodes().toArray(new String[] {});
+ : appCtx.getClusterStateManager().getParticipantNodes().toArray(new String[] {});
for (int i = 0, k = 0; i < count; i++, k = (k + 1) % availableLocations.length) {
chosenLocations.add(availableLocations[k]);
}
@@ -84,6 +85,7 @@
@Override
public void configure(IServiceContext serviceCtx, Map<String, String> configuration) {
+ this.serviceCtx = serviceCtx;
this.configuration = configuration;
}
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/FeedUtils.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/FeedUtils.java
index ec7de91..dad0d51 100644
--- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/FeedUtils.java
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/FeedUtils.java
@@ -26,9 +26,9 @@
import org.apache.asterix.common.cluster.ClusterPartition;
import org.apache.asterix.common.config.ClusterProperties;
+import org.apache.asterix.common.dataflow.ICcApplicationContext;
import org.apache.asterix.common.exceptions.AsterixException;
import org.apache.asterix.common.utils.StoragePathUtil;
-import org.apache.asterix.runtime.utils.ClusterStateManager;
import org.apache.hyracks.algebricks.common.constraints.AlgebricksAbsolutePartitionConstraint;
import org.apache.hyracks.algebricks.common.constraints.AlgebricksPartitionConstraint;
import org.apache.hyracks.algebricks.common.constraints.AlgebricksPartitionConstraint.PartitionConstraintType;
@@ -63,9 +63,9 @@
}
public enum Mode {
- PROCESS, // There is memory
- SPILL, // Memory budget has been consumed. Now we're writing to disk
- DISCARD // Memory and Disk space budgets have been consumed. Now we're discarding
+ PROCESS, // There is memory
+ SPILL, // Memory budget has been consumed. Now we're writing to disk
+ DISCARD // Memory and Disk space budgets have been consumed. Now we're discarding
}
private FeedUtils() {
@@ -87,7 +87,7 @@
return StoragePathUtil.getFileSplitForClusterPartition(partition, f.getPath());
}
- public static FileSplit[] splitsForAdapter(String dataverseName, String feedName,
+ public static FileSplit[] splitsForAdapter(ICcApplicationContext appCtx, String dataverseName, String feedName,
AlgebricksPartitionConstraint partitionConstraints) throws AsterixException {
if (partitionConstraints.getPartitionConstraintType() == PartitionConstraintType.COUNT) {
throw new AsterixException("Can't create file splits for adapter with count partitioning constraints");
@@ -96,7 +96,7 @@
List<FileSplit> splits = new ArrayList<>();
for (String nd : locations) {
splits.add(splitsForAdapter(dataverseName, feedName, nd,
- ClusterStateManager.INSTANCE.getNodePartitions(nd)[0]));
+ appCtx.getClusterStateManager().getNodePartitions(nd)[0]));
}
return splits.toArray(new FileSplit[] {});
}
@@ -113,8 +113,8 @@
public static FeedLogManager getFeedLogManager(IHyracksTaskContext ctx, FileSplit feedLogFileSplit)
throws HyracksDataException {
- return new FeedLogManager(FeedUtils.getAbsoluteFileRef(feedLogFileSplit.getPath(),
- 0, ctx.getIoManager()).getFile());
+ return new FeedLogManager(
+ FeedUtils.getAbsoluteFileRef(feedLogFileSplit.getPath(), 0, ctx.getIoManager()).getFile());
}
public static void processFeedMessage(ByteBuffer input, VSizeFrame message, FrameTupleAccessor fta)
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/HDFSUtils.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/HDFSUtils.java
index b4353e7..bd50352 100644
--- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/HDFSUtils.java
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/HDFSUtils.java
@@ -25,7 +25,9 @@
import java.util.Map;
import org.apache.asterix.common.api.IApplicationContext;
+import org.apache.asterix.common.cluster.IClusterStateManager;
import org.apache.asterix.common.config.DatasetConfig.ExternalFilePendingOp;
+import org.apache.asterix.common.dataflow.ICcApplicationContext;
import org.apache.asterix.common.exceptions.ErrorCode;
import org.apache.asterix.common.exceptions.RuntimeDataException;
import org.apache.asterix.external.indexing.ExternalFile;
@@ -33,7 +35,6 @@
import org.apache.asterix.external.indexing.RecordId.RecordIdType;
import org.apache.asterix.external.input.stream.HDFSInputStream;
import org.apache.asterix.hivecompat.io.RCFileInputFormat;
-import org.apache.asterix.runtime.utils.ClusterStateManager;
import org.apache.hadoop.fs.BlockLocation;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
@@ -207,10 +208,12 @@
public static AlgebricksAbsolutePartitionConstraint getPartitionConstraints(IApplicationContext appCtx,
AlgebricksAbsolutePartitionConstraint clusterLocations) {
if (clusterLocations == null) {
+ IClusterStateManager clusterStateManager = ((ICcApplicationContext) appCtx).getClusterStateManager();
ArrayList<String> locs = new ArrayList<>();
Map<String, String[]> stores = appCtx.getMetadataProperties().getStores();
for (String node : stores.keySet()) {
- int numIODevices = ClusterStateManager.INSTANCE.getIODevices(node).length;
+
+ int numIODevices = clusterStateManager.getIODevices(node).length;
for (int k = 0; k < numIODevices; k++) {
locs.add(node);
}
diff --git a/asterixdb/asterix-external-data/src/test/java/org/apache/asterix/external/input/record/reader/RecordWithPKTestReaderFactory.java b/asterixdb/asterix-external-data/src/test/java/org/apache/asterix/external/input/record/reader/RecordWithPKTestReaderFactory.java
index c9e37ee..715ad02 100644
--- a/asterixdb/asterix-external-data/src/test/java/org/apache/asterix/external/input/record/reader/RecordWithPKTestReaderFactory.java
+++ b/asterixdb/asterix-external-data/src/test/java/org/apache/asterix/external/input/record/reader/RecordWithPKTestReaderFactory.java
@@ -23,7 +23,7 @@
import java.util.List;
import java.util.Map;
-import org.apache.asterix.common.api.IApplicationContext;
+import org.apache.asterix.common.dataflow.ICcApplicationContext;
import org.apache.asterix.external.api.IExternalDataSourceFactory;
import org.apache.asterix.external.api.IRecordReader;
import org.apache.asterix.external.api.IRecordReaderFactory;
@@ -40,11 +40,10 @@
private transient IServiceContext serviceCtx;
private static final List<String> recordReaderNames = Collections.unmodifiableList(Arrays.asList());
-
@Override
public AlgebricksAbsolutePartitionConstraint getPartitionConstraint() throws AlgebricksException {
- clusterLocations = IExternalDataSourceFactory
- .getPartitionConstraints((IApplicationContext) serviceCtx.getApplicationContext(), clusterLocations, 1);
+ clusterLocations = IExternalDataSourceFactory.getPartitionConstraints(
+ (ICcApplicationContext) serviceCtx.getApplicationContext(), clusterLocations, 1);
return clusterLocations;
}
@@ -64,7 +63,8 @@
return RecordWithPK.class;
}
- @Override public List<String> getRecordReaderNames() {
+ @Override
+ public List<String> getRecordReaderNames() {
return recordReaderNames;
}
}
diff --git a/asterixdb/asterix-external-data/src/test/java/org/apache/asterix/external/input/record/reader/kv/KVTestReaderFactory.java b/asterixdb/asterix-external-data/src/test/java/org/apache/asterix/external/input/record/reader/kv/KVTestReaderFactory.java
index c45941d..c09b9eb 100644
--- a/asterixdb/asterix-external-data/src/test/java/org/apache/asterix/external/input/record/reader/kv/KVTestReaderFactory.java
+++ b/asterixdb/asterix-external-data/src/test/java/org/apache/asterix/external/input/record/reader/kv/KVTestReaderFactory.java
@@ -23,9 +23,9 @@
import java.util.List;
import java.util.Map;
+import org.apache.asterix.common.dataflow.ICcApplicationContext;
import org.apache.asterix.external.api.IRecordReader;
import org.apache.asterix.external.api.IRecordReaderFactory;
-import org.apache.asterix.runtime.utils.ClusterStateManager;
import org.apache.hyracks.algebricks.common.constraints.AlgebricksAbsolutePartitionConstraint;
import org.apache.hyracks.api.application.IServiceContext;
import org.apache.hyracks.api.context.IHyracksTaskContext;
@@ -43,17 +43,23 @@
private int upsertCycle = 0;
private int numOfReaders;
private transient AlgebricksAbsolutePartitionConstraint clusterLocations;
+ private transient IServiceContext serviceCtx;
private static final List<String> recordReaderNames = Collections.unmodifiableList(Arrays.asList());
@Override
public AlgebricksAbsolutePartitionConstraint getPartitionConstraint() {
- clusterLocations = ClusterStateManager.INSTANCE.getClusterLocations();
- numOfReaders = clusterLocations.getLocations().length;
+ if (clusterLocations == null) {
+ ICcApplicationContext appCtx = (ICcApplicationContext) serviceCtx.getApplicationContext();
+ clusterLocations = appCtx.getClusterStateManager().getClusterLocations();
+ numOfReaders = clusterLocations.getLocations().length;
+ }
return clusterLocations;
+
}
@Override
public void configure(IServiceContext serviceCtx, final Map<String, String> configuration) {
+ this.serviceCtx = serviceCtx;
if (configuration.containsKey("num-of-records")) {
numOfRecords = Integer.parseInt(configuration.get("num-of-records"));
}
@@ -83,7 +89,8 @@
return DCPRequest.class;
}
- @Override public List<String> getRecordReaderNames() {
+ @Override
+ public List<String> getRecordReaderNames() {
return recordReaderNames;
}
}
diff --git a/asterixdb/asterix-external-data/src/test/java/org/apache/asterix/external/library/adapter/TestTypedAdapterFactory.java b/asterixdb/asterix-external-data/src/test/java/org/apache/asterix/external/library/adapter/TestTypedAdapterFactory.java
index 5262e1f..616ed6e 100644
--- a/asterixdb/asterix-external-data/src/test/java/org/apache/asterix/external/library/adapter/TestTypedAdapterFactory.java
+++ b/asterixdb/asterix-external-data/src/test/java/org/apache/asterix/external/library/adapter/TestTypedAdapterFactory.java
@@ -24,6 +24,7 @@
import org.apache.asterix.common.api.IApplicationContext;
import org.apache.asterix.common.cluster.ClusterPartition;
+import org.apache.asterix.common.dataflow.ICcApplicationContext;
import org.apache.asterix.external.api.IAdapterFactory;
import org.apache.asterix.external.api.IDataSourceAdapter;
import org.apache.asterix.external.api.IExternalDataSourceFactory;
@@ -65,7 +66,7 @@
@Override
public AlgebricksAbsolutePartitionConstraint getPartitionConstraint() throws AlgebricksException {
clusterLocations = IExternalDataSourceFactory.getPartitionConstraints(
- (IApplicationContext) serviceContext.getApplicationContext(), clusterLocations, 1);
+ (ICcApplicationContext) serviceContext.getApplicationContext(), clusterLocations, 1);
return clusterLocations;
}
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 709f655..f5cfbb3 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,7 +22,6 @@
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;
/**
@@ -113,7 +112,7 @@
if (intValue < 0) {
return new Pair<>(false, "Value must be >= 0");
}
- int numNodesInCluster = ClusterStateManager.INSTANCE.getParticipantNodes(true).size();
+ int numNodesInCluster = appCtx.getClusterStateManager().getParticipantNodes(true).size();
if (numNodesInCluster < intValue) {
return new Pair<>(false,
"Value must be less than or equal to the available number of nodes in cluster ("
diff --git a/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/declared/FeedDataSource.java b/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/declared/FeedDataSource.java
index 26cec1e..0b6608c 100644
--- a/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/declared/FeedDataSource.java
+++ b/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/declared/FeedDataSource.java
@@ -22,7 +22,7 @@
import java.util.List;
import org.apache.asterix.active.EntityId;
-import org.apache.asterix.external.feed.api.IFeed;
+import org.apache.asterix.common.dataflow.ICcApplicationContext;
import org.apache.asterix.external.feed.management.FeedConnectionId;
import org.apache.asterix.external.operators.FeedCollectOperatorDescriptor;
import org.apache.asterix.external.util.FeedUtils.FeedRuntimeType;
@@ -34,7 +34,6 @@
import org.apache.asterix.om.types.ARecordType;
import org.apache.asterix.om.types.IAType;
import org.apache.asterix.runtime.formats.NonTaggedDataFormat;
-import org.apache.asterix.runtime.utils.ClusterStateManager;
import org.apache.hyracks.algebricks.common.constraints.AlgebricksAbsolutePartitionConstraint;
import org.apache.hyracks.algebricks.common.constraints.AlgebricksPartitionConstraint;
import org.apache.hyracks.algebricks.common.exceptions.AlgebricksException;
@@ -63,12 +62,12 @@
private final List<ScalarFunctionCallExpression> keyAccessExpression;
private final FeedConnection feedConnection;
- public FeedDataSource(Feed feed, DataSourceId id, String targetDataset, IAType itemType, IAType metaType,
- List<IAType> pkTypes, List<List<String>> partitioningKeys,
- List<ScalarFunctionCallExpression> keyAccessExpression, EntityId sourceFeedId,
- FeedRuntimeType location, String[] locations, INodeDomain domain, FeedConnection feedConnection)
- throws AlgebricksException {
+ public FeedDataSource(MetadataProvider metadataProvider, Feed feed, DataSourceId id, String targetDataset,
+ IAType itemType, IAType metaType, List<IAType> pkTypes,
+ List<ScalarFunctionCallExpression> keyAccessExpression, EntityId sourceFeedId, FeedRuntimeType location,
+ String[] locations, INodeDomain domain, FeedConnection feedConnection) throws AlgebricksException {
super(id, itemType, metaType, Type.FEED, domain);
+ ICcApplicationContext appCtx = metadataProvider.getApplicationContext();
this.feed = feed;
this.targetDataset = targetDataset;
this.sourceFeedId = sourceFeedId;
@@ -76,7 +75,7 @@
this.locations = locations;
this.pkTypes = pkTypes;
this.keyAccessExpression = keyAccessExpression;
- this.computeCardinality = ClusterStateManager.INSTANCE.getParticipantNodes().size();
+ this.computeCardinality = appCtx.getClusterStateManager().getParticipantNodes().size();
this.feedConnection = feedConnection;
initFeedDataSource();
}
@@ -170,8 +169,8 @@
throws AlgebricksException {
try {
ARecordType feedOutputType = (ARecordType) itemType;
- ISerializerDeserializer payloadSerde = NonTaggedDataFormat.INSTANCE.getSerdeProvider()
- .getSerializerDeserializer(feedOutputType);
+ ISerializerDeserializer payloadSerde =
+ NonTaggedDataFormat.INSTANCE.getSerdeProvider().getSerializerDeserializer(feedOutputType);
ArrayList<ISerializerDeserializer> serdes = new ArrayList<>();
serdes.add(payloadSerde);
if (metaItemType != null) {
@@ -182,16 +181,16 @@
serdes.add(SerializerDeserializerProvider.INSTANCE.getSerializerDeserializer(type));
}
}
- RecordDescriptor feedDesc = new RecordDescriptor(
- serdes.toArray(new ISerializerDeserializer[serdes.size()]));
- FeedPolicyEntity feedPolicy = (FeedPolicyEntity) getProperties()
- .get(BuiltinFeedPolicies.CONFIG_FEED_POLICY_KEY);
+ RecordDescriptor feedDesc =
+ new RecordDescriptor(serdes.toArray(new ISerializerDeserializer[serdes.size()]));
+ FeedPolicyEntity feedPolicy =
+ (FeedPolicyEntity) getProperties().get(BuiltinFeedPolicies.CONFIG_FEED_POLICY_KEY);
if (feedPolicy == null) {
throw new AlgebricksException("Feed not configured with a policy");
}
feedPolicy.getProperties().put(BuiltinFeedPolicies.CONFIG_FEED_POLICY_KEY, feedPolicy.getPolicyName());
- FeedConnectionId feedConnectionId = new FeedConnectionId(getId().getDataverseName(),
- getId().getDatasourceName(), getTargetDataset());
+ FeedConnectionId feedConnectionId =
+ new FeedConnectionId(getId().getDataverseName(), getId().getDatasourceName(), getTargetDataset());
FeedCollectOperatorDescriptor feedCollector = new FeedCollectOperatorDescriptor(jobSpec, feedConnectionId,
feedOutputType, feedDesc, feedPolicy.getProperties(), getLocation());
diff --git a/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/declared/MetadataManagerUtil.java b/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/declared/MetadataManagerUtil.java
index c23755d..97c6ed2 100644
--- a/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/declared/MetadataManagerUtil.java
+++ b/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/declared/MetadataManagerUtil.java
@@ -21,6 +21,7 @@
import java.util.ArrayList;
import java.util.List;
+import org.apache.asterix.common.cluster.IClusterStateManager;
import org.apache.asterix.common.config.DatasetConfig.DatasetType;
import org.apache.asterix.common.exceptions.MetadataException;
import org.apache.asterix.metadata.MetadataManager;
@@ -36,7 +37,6 @@
import org.apache.asterix.metadata.utils.MetadataConstants;
import org.apache.asterix.om.types.ARecordType;
import org.apache.asterix.om.types.IAType;
-import org.apache.asterix.runtime.utils.ClusterStateManager;
import org.apache.hyracks.algebricks.common.exceptions.AlgebricksException;
import org.apache.hyracks.algebricks.core.algebra.properties.DefaultNodeGroupDomain;
import org.apache.hyracks.algebricks.core.algebra.properties.INodeDomain;
@@ -111,12 +111,12 @@
return dataset;
}
- public static INodeDomain findNodeDomain(MetadataTransactionContext mdTxnCtx, String nodeGroupName)
- throws AlgebricksException {
+ public static INodeDomain findNodeDomain(IClusterStateManager clusterStateManager,
+ MetadataTransactionContext mdTxnCtx, String nodeGroupName) throws AlgebricksException {
NodeGroup nodeGroup = MetadataManager.INSTANCE.getNodegroup(mdTxnCtx, nodeGroupName);
List<String> partitions = new ArrayList<>();
for (String location : nodeGroup.getNodeNames()) {
- int numPartitions = ClusterStateManager.INSTANCE.getNodePartitionsCount(location);
+ int numPartitions = clusterStateManager.getNodePartitionsCount(location);
for (int i = 0; i < numPartitions; i++) {
partitions.add(location);
}
@@ -165,24 +165,24 @@
}
}
- public static DataSource findDataSource(MetadataTransactionContext mdTxnCtx, DataSourceId id)
- throws AlgebricksException {
+ public static DataSource findDataSource(IClusterStateManager clusterStateManager,
+ MetadataTransactionContext mdTxnCtx, DataSourceId id) throws AlgebricksException {
try {
- return lookupSourceInMetadata(mdTxnCtx, id);
+ return lookupSourceInMetadata(clusterStateManager, mdTxnCtx, id);
} catch (MetadataException e) {
throw new AlgebricksException(e);
}
}
- public static DataSource lookupSourceInMetadata(MetadataTransactionContext mdTxnCtx, DataSourceId aqlId)
- throws AlgebricksException {
+ public static DataSource lookupSourceInMetadata(IClusterStateManager clusterStateManager,
+ MetadataTransactionContext mdTxnCtx, DataSourceId aqlId) throws AlgebricksException {
Dataset dataset = findDataset(mdTxnCtx, aqlId.getDataverseName(), aqlId.getDatasourceName());
if (dataset == null) {
throw new AlgebricksException("Datasource with id " + aqlId + " was not found.");
}
IAType itemType = findType(mdTxnCtx, dataset.getItemTypeDataverseName(), dataset.getItemTypeName());
IAType metaItemType = findType(mdTxnCtx, dataset.getMetaItemTypeDataverseName(), dataset.getMetaItemTypeName());
- INodeDomain domain = findNodeDomain(mdTxnCtx, dataset.getNodeGroupName());
+ INodeDomain domain = findNodeDomain(clusterStateManager, mdTxnCtx, dataset.getNodeGroupName());
byte datasourceType = dataset.getDatasetType().equals(DatasetType.EXTERNAL) ? DataSource.Type.EXTERNAL_DATASET
: DataSource.Type.INTERNAL_DATASET;
return new DatasetDataSource(aqlId, dataset, itemType, metaItemType, datasourceType,
diff --git a/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/declared/MetadataProvider.java b/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/declared/MetadataProvider.java
index 8971a90..d6a3f21 100644
--- a/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/declared/MetadataProvider.java
+++ b/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/declared/MetadataProvider.java
@@ -26,6 +26,7 @@
import java.util.List;
import java.util.Map;
+import org.apache.asterix.common.cluster.IClusterStateManager;
import org.apache.asterix.common.config.DatasetConfig.DatasetType;
import org.apache.asterix.common.config.DatasetConfig.ExternalFilePendingOp;
import org.apache.asterix.common.config.DatasetConfig.IndexType;
@@ -84,7 +85,6 @@
import org.apache.asterix.runtime.formats.FormatUtils;
import org.apache.asterix.runtime.job.listener.JobEventListenerFactory;
import org.apache.asterix.runtime.operators.LSMSecondaryUpsertOperatorDescriptor;
-import org.apache.asterix.runtime.utils.ClusterStateManager;
import org.apache.hyracks.algebricks.common.constraints.AlgebricksAbsolutePartitionConstraint;
import org.apache.hyracks.algebricks.common.constraints.AlgebricksPartitionConstraint;
import org.apache.hyracks.algebricks.common.exceptions.AlgebricksException;
@@ -295,7 +295,7 @@
}
public INodeDomain findNodeDomain(String nodeGroupName) throws AlgebricksException {
- return MetadataManagerUtil.findNodeDomain(mdTxnCtx, nodeGroupName);
+ return MetadataManagerUtil.findNodeDomain(appCtx.getClusterStateManager(), mdTxnCtx, nodeGroupName);
}
public List<String> findNodes(String nodeGroupName) throws AlgebricksException {
@@ -329,11 +329,11 @@
@Override
public DataSource findDataSource(DataSourceId id) throws AlgebricksException {
- return MetadataManagerUtil.findDataSource(mdTxnCtx, id);
+ return MetadataManagerUtil.findDataSource(appCtx.getClusterStateManager(), mdTxnCtx, id);
}
public DataSource lookupSourceInMetadata(DataSourceId aqlId) throws AlgebricksException {
- return MetadataManagerUtil.lookupSourceInMetadata(mdTxnCtx, aqlId);
+ return MetadataManagerUtil.lookupSourceInMetadata(appCtx.getClusterStateManager(), mdTxnCtx, aqlId);
}
@Override
@@ -709,8 +709,9 @@
int numPartitions = 0;
List<String> nodeGroup =
MetadataManager.INSTANCE.getNodegroup(mdTxnCtx, dataset.getNodeGroupName()).getNodeNames();
+ IClusterStateManager csm = appCtx.getClusterStateManager();
for (String nd : nodeGroup) {
- numPartitions += ClusterStateManager.INSTANCE.getNodePartitionsCount(nd);
+ numPartitions += csm.getNodePartitionsCount(nd);
}
return numElementsHint / numPartitions;
}
@@ -755,12 +756,13 @@
}
public Pair<IFileSplitProvider, AlgebricksPartitionConstraint> splitAndConstraints(String dataverse) {
- return SplitsAndConstraintsUtil.getDataverseSplitProviderAndConstraints(dataverse);
+ return SplitsAndConstraintsUtil.getDataverseSplitProviderAndConstraints(appCtx.getClusterStateManager(),
+ dataverse);
}
public FileSplit[] splitsForIndex(MetadataTransactionContext mdTxnCtx, Dataset dataset, String indexName)
throws AlgebricksException {
- return SplitsAndConstraintsUtil.getIndexSplits(dataset, indexName, mdTxnCtx);
+ return SplitsAndConstraintsUtil.getIndexSplits(appCtx.getClusterStateManager(), dataset, indexName, mdTxnCtx);
}
public DatasourceAdapter getAdapter(MetadataTransactionContext mdTxnCtx, String dataverseName, String adapterName)
@@ -777,7 +779,7 @@
}
public AlgebricksAbsolutePartitionConstraint getClusterLocations() {
- return ClusterStateManager.INSTANCE.getClusterLocations();
+ return appCtx.getClusterStateManager().getClusterLocations();
}
public Pair<IOperatorDescriptor, AlgebricksPartitionConstraint> buildExternalDataLookupRuntime(
diff --git a/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/utils/SplitsAndConstraintsUtil.java b/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/utils/SplitsAndConstraintsUtil.java
index 6825f10..5b7ea59 100644
--- a/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/utils/SplitsAndConstraintsUtil.java
+++ b/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/utils/SplitsAndConstraintsUtil.java
@@ -23,6 +23,7 @@
import java.util.List;
import org.apache.asterix.common.cluster.ClusterPartition;
+import org.apache.asterix.common.cluster.IClusterStateManager;
import org.apache.asterix.common.config.ClusterProperties;
import org.apache.asterix.common.exceptions.MetadataException;
import org.apache.asterix.common.utils.StoragePathUtil;
@@ -30,7 +31,6 @@
import org.apache.asterix.metadata.MetadataTransactionContext;
import org.apache.asterix.metadata.entities.Dataset;
import org.apache.asterix.metadata.entities.NodeGroup;
-import org.apache.asterix.runtime.utils.ClusterStateManager;
import org.apache.hyracks.algebricks.common.constraints.AlgebricksPartitionConstraint;
import org.apache.hyracks.algebricks.common.exceptions.AlgebricksException;
import org.apache.hyracks.algebricks.common.utils.Pair;
@@ -42,11 +42,11 @@
private SplitsAndConstraintsUtil() {
}
- private static FileSplit[] getDataverseSplits(String dataverseName) {
+ private static FileSplit[] getDataverseSplits(IClusterStateManager clusterStateManager, String dataverseName) {
File relPathFile = new File(dataverseName);
List<FileSplit> splits = new ArrayList<>();
// get all partitions
- ClusterPartition[] clusterPartition = ClusterStateManager.INSTANCE.getClusterPartitons();
+ ClusterPartition[] clusterPartition = clusterStateManager.getClusterPartitons();
String storageDirName = ClusterProperties.INSTANCE.getStorageDirectoryName();
for (int j = 0; j < clusterPartition.length; j++) {
File f = new File(
@@ -57,28 +57,29 @@
return splits.toArray(new FileSplit[] {});
}
- public static FileSplit[] getIndexSplits(Dataset dataset, String indexName, MetadataTransactionContext mdTxnCtx)
- throws AlgebricksException {
+ public static FileSplit[] getIndexSplits(IClusterStateManager clusterStateManager, Dataset dataset,
+ String indexName, MetadataTransactionContext mdTxnCtx) throws AlgebricksException {
try {
NodeGroup nodeGroup = MetadataManager.INSTANCE.getNodegroup(mdTxnCtx, dataset.getNodeGroupName());
if (nodeGroup == null) {
throw new AlgebricksException("Couldn't find node group " + dataset.getNodeGroupName());
}
List<String> nodeList = nodeGroup.getNodeNames();
- return getIndexSplits(dataset, indexName, nodeList);
+ return getIndexSplits(clusterStateManager, dataset, indexName, nodeList);
} catch (MetadataException me) {
throw new AlgebricksException(me);
}
}
- public static FileSplit[] getIndexSplits(Dataset dataset, String indexName, List<String> nodes) {
+ public static FileSplit[] getIndexSplits(IClusterStateManager clusterStateManager, Dataset dataset,
+ String indexName, List<String> nodes) {
File relPathFile = new File(StoragePathUtil.prepareDataverseIndexName(dataset.getDataverseName(),
dataset.getDatasetName(), indexName, dataset.getRebalanceCount()));
String storageDirName = ClusterProperties.INSTANCE.getStorageDirectoryName();
List<FileSplit> splits = new ArrayList<>();
for (String nd : nodes) {
- int numPartitions = ClusterStateManager.INSTANCE.getNodePartitionsCount(nd);
- ClusterPartition[] nodePartitions = ClusterStateManager.INSTANCE.getNodePartitions(nd);
+ int numPartitions = clusterStateManager.getNodePartitionsCount(nd);
+ ClusterPartition[] nodePartitions = clusterStateManager.getNodePartitions(nd);
// currently this case is never executed since the metadata group doesn't exists
if (dataset.getNodeGroupName().compareTo(MetadataConstants.METADATA_NODEGROUP_NAME) == 0) {
numPartitions = 1;
@@ -97,8 +98,8 @@
}
public static Pair<IFileSplitProvider, AlgebricksPartitionConstraint> getDataverseSplitProviderAndConstraints(
- String dataverse) {
- FileSplit[] splits = getDataverseSplits(dataverse);
+ IClusterStateManager clusterStateManager, String dataverse) {
+ FileSplit[] splits = getDataverseSplits(clusterStateManager, dataverse);
return StoragePathUtil.splitProviderAndPartitionConstraints(splits);
}
diff --git a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/message/ResourceIdRequestMessage.java b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/message/ResourceIdRequestMessage.java
index 194fd59..82a1177 100644
--- a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/message/ResourceIdRequestMessage.java
+++ b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/message/ResourceIdRequestMessage.java
@@ -20,11 +20,11 @@
import java.util.Set;
+import org.apache.asterix.common.cluster.IClusterStateManager;
import org.apache.asterix.common.dataflow.ICcApplicationContext;
import org.apache.asterix.common.messaging.api.ICCMessageBroker;
import org.apache.asterix.common.messaging.api.ICcAddressedMessage;
import org.apache.asterix.common.transactions.IResourceIdManager;
-import org.apache.asterix.runtime.utils.ClusterStateManager;
import org.apache.hyracks.api.exceptions.HyracksDataException;
public class ResourceIdRequestMessage implements ICcAddressedMessage {
@@ -40,7 +40,8 @@
try {
ICCMessageBroker broker = (ICCMessageBroker) appCtx.getServiceContext().getMessageBroker();
ResourceIdRequestResponseMessage reponse = new ResourceIdRequestResponseMessage();
- if (!ClusterStateManager.INSTANCE.isClusterActive()) {
+ IClusterStateManager clusterStateManager = appCtx.getClusterStateManager();
+ if (!clusterStateManager.isClusterActive()) {
reponse.setResourceId(-1);
reponse.setException(new Exception("Cannot generate global resource id when cluster is not active."));
} else {
@@ -49,7 +50,7 @@
if (reponse.getResourceId() < 0) {
reponse.setException(new Exception("One or more nodes has not reported max resource id."));
}
- requestMaxResourceID(resourceIdManager, broker);
+ requestMaxResourceID(clusterStateManager, resourceIdManager, broker);
}
broker.sendApplicationMessageToNC(reponse, src);
} catch (Exception e) {
@@ -57,8 +58,9 @@
}
}
- private void requestMaxResourceID(IResourceIdManager resourceIdManager, ICCMessageBroker broker) throws Exception {
- Set<String> getParticipantNodes = ClusterStateManager.INSTANCE.getParticipantNodes();
+ private void requestMaxResourceID(IClusterStateManager clusterStateManager, IResourceIdManager resourceIdManager,
+ ICCMessageBroker broker) throws Exception {
+ Set<String> getParticipantNodes = clusterStateManager.getParticipantNodes();
ReportMaxResourceIdRequestMessage msg = new ReportMaxResourceIdRequestMessage();
for (String nodeId : getParticipantNodes) {
if (!resourceIdManager.reported(nodeId)) {
diff --git a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/cc/ResourceIdManager.java b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/transaction/ResourceIdManager.java
similarity index 84%
rename from asterixdb/asterix-app/src/main/java/org/apache/asterix/app/cc/ResourceIdManager.java
rename to asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/transaction/ResourceIdManager.java
index 372404c..6a5ed08 100644
--- a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/cc/ResourceIdManager.java
+++ b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/transaction/ResourceIdManager.java
@@ -16,27 +16,32 @@
* specific language governing permissions and limitations
* under the License.
*/
-package org.apache.asterix.app.cc;
+package org.apache.asterix.runtime.transaction;
import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.atomic.AtomicLong;
+import org.apache.asterix.common.cluster.IClusterStateManager;
import org.apache.asterix.common.transactions.IResourceIdManager;
-import org.apache.asterix.runtime.utils.ClusterStateManager;
public class ResourceIdManager implements IResourceIdManager {
+ private final IClusterStateManager csm;
private final AtomicLong globalResourceId = new AtomicLong();
private volatile Set<String> reportedNodes = new HashSet<>();
private volatile boolean allReported = false;
+ public ResourceIdManager(IClusterStateManager csm) {
+ this.csm = csm;
+ }
+
@Override
public long createResourceId() {
if (!allReported) {
synchronized (this) {
if (!allReported) {
- if (reportedNodes.size() < ClusterStateManager.INSTANCE.getNumberOfNodes()) {
+ if (reportedNodes.size() < csm.getNumberOfNodes()) {
return -1;
} else {
reportedNodes = null;
@@ -58,7 +63,7 @@
if (!allReported) {
globalResourceId.set(Math.max(maxResourceId, globalResourceId.get()));
reportedNodes.add(nodeId);
- if (reportedNodes.size() == ClusterStateManager.INSTANCE.getNumberOfNodes()) {
+ if (reportedNodes.size() == csm.getNumberOfNodes()) {
reportedNodes = null;
allReported = true;
}
diff --git a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/utils/CcApplicationContext.java b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/utils/CcApplicationContext.java
index 28c480f..e4cc7f4 100644
--- a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/utils/CcApplicationContext.java
+++ b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/utils/CcApplicationContext.java
@@ -43,6 +43,7 @@
import org.apache.asterix.common.metadata.IMetadataBootstrap;
import org.apache.asterix.common.replication.IFaultToleranceStrategy;
import org.apache.asterix.common.transactions.IResourceIdManager;
+import org.apache.asterix.runtime.transaction.ResourceIdManager;
import org.apache.hyracks.api.application.ICCServiceContext;
import org.apache.hyracks.api.client.IHyracksClientConnection;
import org.apache.hyracks.api.job.IJobLifecycleListener;
@@ -77,17 +78,16 @@
private IFaultToleranceStrategy ftStrategy;
private IJobLifecycleListener activeLifeCycleListener;
private IMetadataLockManager mdLockManager;
+ private IClusterStateManager clusterStateManager;
public CcApplicationContext(ICCServiceContext ccServiceCtx, IHyracksClientConnection hcc,
- ILibraryManager libraryManager, IResourceIdManager resourceIdManager,
- Supplier<IMetadataBootstrap> metadataBootstrapSupplier, IGlobalRecoveryManager globalRecoveryManager,
- IFaultToleranceStrategy ftStrategy, IJobLifecycleListener activeLifeCycleListener,
- IStorageComponentProvider storageComponentProvider, IMetadataLockManager mdLockManager)
- throws AsterixException, IOException {
+ ILibraryManager libraryManager, Supplier<IMetadataBootstrap> metadataBootstrapSupplier,
+ IGlobalRecoveryManager globalRecoveryManager, IFaultToleranceStrategy ftStrategy,
+ IJobLifecycleListener activeLifeCycleListener, IStorageComponentProvider storageComponentProvider,
+ IMetadataLockManager mdLockManager) throws AsterixException, IOException {
this.ccServiceCtx = ccServiceCtx;
this.hcc = hcc;
this.libraryManager = libraryManager;
- this.resourceIdManager = resourceIdManager;
this.activeLifeCycleListener = activeLifeCycleListener;
// Determine whether to use old-style asterix-configuration.xml or new-style configuration.
// QQQ strip this out eventually
@@ -109,6 +109,9 @@
this.globalRecoveryManager = globalRecoveryManager;
this.storageComponentProvider = storageComponentProvider;
this.mdLockManager = mdLockManager;
+ clusterStateManager = new ClusterStateManager();
+ clusterStateManager.setCcAppCtx(this);
+ this.resourceIdManager = new ResourceIdManager(clusterStateManager);
}
@Override
@@ -204,6 +207,7 @@
return resourceIdManager;
}
+ @Override
public IMetadataBootstrap getMetadataBootstrap() {
return metadataBootstrapSupplier.get();
}
@@ -230,6 +234,6 @@
@Override
public IClusterStateManager getClusterStateManager() {
- return ClusterStateManager.INSTANCE;
+ return clusterStateManager;
}
}
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 cdb3112..36cb10d 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
@@ -36,6 +36,7 @@
import org.apache.asterix.common.cluster.ClusterPartition;
import org.apache.asterix.common.cluster.IClusterStateManager;
import org.apache.asterix.common.config.ClusterProperties;
+import org.apache.asterix.common.dataflow.ICcApplicationContext;
import org.apache.asterix.common.exceptions.AsterixException;
import org.apache.asterix.common.exceptions.ErrorCode;
import org.apache.asterix.common.replication.IFaultToleranceStrategy;
@@ -66,7 +67,6 @@
*/
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;
@@ -78,13 +78,14 @@
private boolean metadataNodeActive = false;
private Set<String> failedNodes = new HashSet<>();
private IFaultToleranceStrategy ftStrategy;
- private CcApplicationContext appCtx;
+ private ICcApplicationContext appCtx;
- private ClusterStateManager() {
+ public ClusterStateManager() {
cluster = ClusterProperties.INSTANCE.getCluster();
}
- public void setCcAppCtx(CcApplicationContext appCtx) {
+ @Override
+ public void setCcAppCtx(ICcApplicationContext appCtx) {
this.appCtx = appCtx;
node2PartitionsMap = appCtx.getMetadataProperties().getNodePartitions();
clusterPartitions = appCtx.getMetadataProperties().getClusterPartitions();
@@ -93,6 +94,7 @@
ftStrategy.bindTo(this);
}
+ @Override
public synchronized void removeNCConfiguration(String nodeId) throws HyracksException {
if (LOGGER.isLoggable(Level.INFO)) {
LOGGER.info("Removing configuration parameters for node id " + nodeId);
@@ -102,6 +104,7 @@
pendingRemoval.remove(nodeId);
}
+ @Override
public synchronized void addNCConfiguration(String nodeId, Map<IOption, Object> configuration)
throws HyracksException {
if (LOGGER.isLoggable(Level.INFO)) {
@@ -209,13 +212,7 @@
return true;
}
- /**
- * Returns the IO devices configured for a Node Controller
- *
- * @param nodeId
- * unique identifier of the Node Controller
- * @return a list of IO devices.
- */
+ @Override
public synchronized String[] getIODevices(String nodeId) {
Map<IOption, Object> ncConfig = activeNcConfiguration.get(nodeId);
if (ncConfig == null) {
@@ -233,11 +230,13 @@
return state;
}
+ @Override
public synchronized Node getAvailableSubstitutionNode() {
List<Node> subNodes = cluster.getSubstituteNodes() == null ? null : cluster.getSubstituteNodes().getNode();
return subNodes == null || subNodes.isEmpty() ? null : subNodes.get(0);
}
+ @Override
public synchronized Set<String> getParticipantNodes() {
Set<String> participantNodes = new HashSet<>();
for (String pNode : activeNcConfiguration.keySet()) {
@@ -246,6 +245,7 @@
return participantNodes;
}
+ @Override
public synchronized Set<String> getParticipantNodes(boolean excludePendingRemoval) {
Set<String> participantNodes = getParticipantNodes();
if (excludePendingRemoval) {
@@ -254,6 +254,7 @@
return participantNodes;
}
+ @Override
public synchronized AlgebricksAbsolutePartitionConstraint getClusterLocations() {
if (clusterPartitionConstraint == null) {
resetClusterPartitionConstraint();
@@ -272,6 +273,7 @@
new AlgebricksAbsolutePartitionConstraint(clusterActiveLocations.toArray(new String[] {}));
}
+ @Override
public synchronized boolean isClusterActive() {
if (cluster == null) {
// this is a virtual cluster
@@ -280,6 +282,7 @@
return state == ClusterState.ACTIVE;
}
+ @Override
public int getNumberOfNodes() {
return appCtx.getMetadataProperties().getNodeNames().size();
}
@@ -289,6 +292,7 @@
return node2PartitionsMap.get(nodeId);
}
+ @Override
public synchronized int getNodePartitionsCount(String node) {
if (node2PartitionsMap.containsKey(node)) {
return node2PartitionsMap.get(node).length;
@@ -305,10 +309,12 @@
return partitons.toArray(new ClusterPartition[] {});
}
+ @Override
public synchronized boolean isMetadataNodeActive() {
return metadataNodeActive;
}
+ @Override
public synchronized ObjectNode getClusterStateDescription() {
ObjectMapper om = new ObjectMapper();
ObjectNode stateDescription = om.createObjectNode();
@@ -342,6 +348,7 @@
return stateDescription;
}
+ @Override
public synchronized ObjectNode getClusterStateSummary() {
ObjectMapper om = new ObjectMapper();
ObjectNode stateDescription = om.createObjectNode();
@@ -395,6 +402,7 @@
}
}
+ @Override
public synchronized void removePending(String nodeId) {
if (LOGGER.isLoggable(Level.INFO)) {
LOGGER.info("Registering intention to remove node id " + nodeId);
@@ -406,6 +414,7 @@
}
}
+ @Override
public synchronized boolean cancelRemovePending(String nodeId) {
if (LOGGER.isLoggable(Level.INFO)) {
LOGGER.info("Deregistering intention to remove node id " + nodeId);
@@ -423,8 +432,8 @@
}
private void updateNodeConfig(String nodeId, Map<IOption, Object> configuration) {
- ConfigManager configManager = ((ConfigManagerApplicationConfig) appCtx.getServiceContext().getAppConfig())
- .getConfigManager();
+ ConfigManager configManager =
+ ((ConfigManagerApplicationConfig) appCtx.getServiceContext().getAppConfig()).getConfigManager();
for (Map.Entry<IOption, Object> entry : configuration.entrySet()) {
if (entry.getKey().section() == Section.NC) {
configManager.set(nodeId, entry.getKey(), entry.getValue());