You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@asterixdb.apache.org by am...@apache.org on 2017/03/10 08:14:24 UTC
[5/6] asterixdb git commit: Unify structure of Cluster Controller and
Node Controller
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/953eb966/asterixdb/asterix-app/src/main/java/org/apache/asterix/hyracks/bootstrap/CCApplication.java
----------------------------------------------------------------------
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
new file mode 100644
index 0000000..dadc2ae
--- /dev/null
+++ b/asterixdb/asterix-app/src/main/java/org/apache/asterix/hyracks/bootstrap/CCApplication.java
@@ -0,0 +1,325 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.asterix.hyracks.bootstrap;
+
+import static org.apache.asterix.api.http.servlet.ServletConstants.ASTERIX_APP_CONTEXT_INFO_ATTR;
+import static org.apache.asterix.api.http.servlet.ServletConstants.HYRACKS_CONNECTION_ATTR;
+
+import java.util.Arrays;
+import java.util.List;
+import java.util.concurrent.ConcurrentMap;
+import java.util.logging.Level;
+import java.util.logging.Logger;
+
+import org.apache.asterix.active.ActiveLifecycleListener;
+import org.apache.asterix.api.http.server.ApiServlet;
+import org.apache.asterix.api.http.server.ClusterApiServlet;
+import org.apache.asterix.api.http.server.ClusterControllerDetailsApiServlet;
+import org.apache.asterix.api.http.server.ConnectorApiServlet;
+import org.apache.asterix.api.http.server.DdlApiServlet;
+import org.apache.asterix.api.http.server.DiagnosticsApiServlet;
+import org.apache.asterix.api.http.server.FeedServlet;
+import org.apache.asterix.api.http.server.FullApiServlet;
+import org.apache.asterix.api.http.server.NodeControllerDetailsApiServlet;
+import org.apache.asterix.api.http.server.QueryApiServlet;
+import org.apache.asterix.api.http.server.QueryResultApiServlet;
+import org.apache.asterix.api.http.server.QueryServiceServlet;
+import org.apache.asterix.api.http.server.QueryStatusApiServlet;
+import org.apache.asterix.api.http.server.QueryWebInterfaceServlet;
+import org.apache.asterix.api.http.server.ShutdownApiServlet;
+import org.apache.asterix.api.http.server.UpdateApiServlet;
+import org.apache.asterix.api.http.server.VersionApiServlet;
+import org.apache.asterix.api.http.servlet.ServletConstants;
+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;
+import org.apache.asterix.common.config.AsterixExtension;
+import org.apache.asterix.common.config.ClusterProperties;
+import org.apache.asterix.common.config.ExternalProperties;
+import org.apache.asterix.common.config.MetadataProperties;
+import org.apache.asterix.common.context.IStorageComponentProvider;
+import org.apache.asterix.common.library.ILibraryManager;
+import org.apache.asterix.common.messaging.api.ICCMessageBroker;
+import org.apache.asterix.common.replication.IFaultToleranceStrategy;
+import org.apache.asterix.common.replication.IReplicationStrategy;
+import org.apache.asterix.common.utils.Servlets;
+import org.apache.asterix.external.library.ExternalLibraryManager;
+import org.apache.asterix.file.StorageComponentProvider;
+import org.apache.asterix.messaging.CCMessageBroker;
+import org.apache.asterix.metadata.MetadataManager;
+import org.apache.asterix.metadata.api.IAsterixStateProxy;
+import org.apache.asterix.metadata.bootstrap.AsterixStateProxy;
+import org.apache.asterix.metadata.cluster.ClusterManagerProvider;
+import org.apache.asterix.runtime.job.resource.JobCapacityController;
+import org.apache.asterix.runtime.utils.AppContextInfo;
+import org.apache.asterix.translator.IStatementExecutorFactory;
+import org.apache.hyracks.api.application.ICCServiceContext;
+import org.apache.hyracks.api.application.IServiceContext;
+import org.apache.hyracks.api.client.HyracksConnection;
+import org.apache.hyracks.api.client.IHyracksClientConnection;
+import org.apache.hyracks.api.config.IConfigManager;
+import org.apache.hyracks.api.job.resource.IJobCapacityController;
+import org.apache.hyracks.api.lifecycle.LifeCycleComponentManager;
+import org.apache.hyracks.control.cc.BaseCCApplication;
+import org.apache.hyracks.control.cc.ClusterControllerService;
+import org.apache.hyracks.control.common.controllers.CCConfig;
+import org.apache.hyracks.http.api.IServlet;
+import org.apache.hyracks.http.server.HttpServer;
+import org.apache.hyracks.http.server.WebManager;
+
+public class CCApplication extends BaseCCApplication {
+
+ private static final Logger LOGGER = Logger.getLogger(CCApplication.class.getName());
+ private static IAsterixStateProxy proxy;
+ protected ICCServiceContext ccServiceCtx;
+ protected CCExtensionManager ccExtensionManager;
+ protected IStorageComponentProvider componentProvider;
+ private IJobCapacityController jobCapacityController;
+ protected WebManager webManager;
+
+ @Override
+ public void start(IServiceContext serviceCtx, String[] args) throws Exception {
+ if (args.length > 0) {
+ throw new IllegalArgumentException("Unrecognized argument(s): " + Arrays.toString(args));
+ }
+ final ClusterControllerService controllerService = (ClusterControllerService) serviceCtx.getControllerService();
+ ICCMessageBroker messageBroker = new CCMessageBroker(controllerService);
+ this.ccServiceCtx = (ICCServiceContext) serviceCtx;
+
+ if (LOGGER.isLoggable(Level.INFO)) {
+ LOGGER.info("Starting Asterix cluster controller");
+ }
+
+ ccServiceCtx.setThreadFactory(
+ new AsterixThreadFactory(ccServiceCtx.getThreadFactory(), new LifeCycleComponentManager()));
+ ILibraryManager libraryManager = new ExternalLibraryManager();
+ ResourceIdManager resourceIdManager = new ResourceIdManager();
+ IReplicationStrategy repStrategy = ClusterProperties.INSTANCE.getReplicationStrategy();
+ IFaultToleranceStrategy ftStrategy = FaultToleranceStrategyFactory
+ .create(ClusterProperties.INSTANCE.getCluster(), repStrategy, messageBroker);
+ ExternalLibraryUtils.setUpExternaLibraries(libraryManager, false);
+ componentProvider = new StorageComponentProvider();
+ GlobalRecoveryManager.instantiate((HyracksConnection) getHcc(), componentProvider);
+ AppContextInfo.initialize(ccServiceCtx, getHcc(), libraryManager, resourceIdManager,
+ () -> MetadataManager.INSTANCE, GlobalRecoveryManager.instance(), ftStrategy);
+ ccExtensionManager = new CCExtensionManager(getExtensions());
+ AppContextInfo.INSTANCE.setExtensionManager(ccExtensionManager);
+ final CCConfig ccConfig = controllerService.getCCConfig();
+ if (System.getProperty("java.rmi.server.hostname") == null) {
+ System.setProperty("java.rmi.server.hostname", ccConfig.getClusterListenAddress());
+ }
+ MetadataProperties metadataProperties = AppContextInfo.INSTANCE.getMetadataProperties();
+
+ setAsterixStateProxy(AsterixStateProxy.registerRemoteObject(metadataProperties.getMetadataCallbackPort()));
+ ccServiceCtx.setDistributedState(proxy);
+
+ MetadataManager.initialize(proxy, metadataProperties);
+
+ AppContextInfo.INSTANCE.getCCServiceContext().addJobLifecycleListener(ActiveLifecycleListener.INSTANCE);
+
+ // create event loop groups
+ webManager = new WebManager();
+ configureServers();
+ webManager.start();
+ ClusterManagerProvider.getClusterManager().registerSubscriber(GlobalRecoveryManager.instance());
+ ccServiceCtx.addClusterLifecycleListener(ClusterLifecycleListener.INSTANCE);
+ ccServiceCtx.setMessageBroker(messageBroker);
+
+ jobCapacityController = new JobCapacityController(controllerService.getResourceManager());
+ }
+
+ protected List<AsterixExtension> getExtensions() {
+ return AppContextInfo.INSTANCE.getExtensionProperties().getExtensions();
+ }
+
+ protected void configureServers() throws Exception {
+ webManager.add(setupWebServer(AppContextInfo.INSTANCE.getExternalProperties()));
+ webManager.add(setupJSONAPIServer(AppContextInfo.INSTANCE.getExternalProperties()));
+ webManager.add(setupFeedServer(AppContextInfo.INSTANCE.getExternalProperties()));
+ webManager.add(setupQueryWebServer(AppContextInfo.INSTANCE.getExternalProperties()));
+ }
+
+ @Override
+ public void stop() throws Exception {
+ ActiveLifecycleListener.INSTANCE.stop();
+ if (LOGGER.isLoggable(Level.INFO)) {
+ LOGGER.info("Stopping Asterix cluster controller");
+ }
+ AsterixStateProxy.unregisterRemoteObject();
+ webManager.stop();
+ }
+
+ protected HttpServer setupWebServer(ExternalProperties externalProperties) throws Exception {
+ HttpServer webServer = new HttpServer(webManager.getBosses(), webManager.getWorkers(),
+ externalProperties.getWebInterfacePort());
+ IHyracksClientConnection hcc = getHcc();
+ webServer.setAttribute(HYRACKS_CONNECTION_ATTR, hcc);
+ webServer.addServlet(new ApiServlet(webServer.ctx(), new String[] { "/*" },
+ ccExtensionManager.getAqlCompilationProvider(), ccExtensionManager.getSqlppCompilationProvider(),
+ getStatementExecutorFactory(), componentProvider));
+ return webServer;
+ }
+
+ protected HttpServer setupJSONAPIServer(ExternalProperties externalProperties) throws Exception {
+ HttpServer jsonAPIServer =
+ new HttpServer(webManager.getBosses(), webManager.getWorkers(), externalProperties.getAPIServerPort());
+ IHyracksClientConnection hcc = getHcc();
+ jsonAPIServer.setAttribute(HYRACKS_CONNECTION_ATTR, hcc);
+ jsonAPIServer.setAttribute(ASTERIX_APP_CONTEXT_INFO_ATTR, AppContextInfo.INSTANCE);
+ jsonAPIServer.setAttribute(ServletConstants.EXECUTOR_SERVICE,
+ ((ClusterControllerService) ccServiceCtx.getControllerService()).getExecutor());
+
+ // AQL rest APIs.
+ addServlet(jsonAPIServer, Servlets.AQL_QUERY);
+ addServlet(jsonAPIServer, Servlets.AQL_UPDATE);
+ addServlet(jsonAPIServer, Servlets.AQL_DDL);
+ addServlet(jsonAPIServer, Servlets.AQL);
+
+ // SQL+x+ rest APIs.
+ addServlet(jsonAPIServer, Servlets.SQLPP_QUERY);
+ addServlet(jsonAPIServer, Servlets.SQLPP_UPDATE);
+ addServlet(jsonAPIServer, Servlets.SQLPP_DDL);
+ addServlet(jsonAPIServer, Servlets.SQLPP);
+
+ // Other APIs.
+ addServlet(jsonAPIServer, Servlets.QUERY_STATUS);
+ addServlet(jsonAPIServer, Servlets.QUERY_RESULT);
+ addServlet(jsonAPIServer, Servlets.QUERY_SERVICE);
+ addServlet(jsonAPIServer, Servlets.CONNECTOR);
+ addServlet(jsonAPIServer, Servlets.SHUTDOWN);
+ addServlet(jsonAPIServer, Servlets.VERSION);
+ addServlet(jsonAPIServer, Servlets.CLUSTER_STATE);
+ addServlet(jsonAPIServer, Servlets.CLUSTER_STATE_NODE_DETAIL); // must not precede add of CLUSTER_STATE
+ addServlet(jsonAPIServer, Servlets.CLUSTER_STATE_CC_DETAIL); // must not precede add of CLUSTER_STATE
+ addServlet(jsonAPIServer, Servlets.DIAGNOSTICS);
+ return jsonAPIServer;
+ }
+
+ protected void addServlet(HttpServer server, String path) {
+ server.addServlet(createServlet(server.ctx(), path, path));
+ }
+
+ protected HttpServer setupQueryWebServer(ExternalProperties externalProperties) throws Exception {
+ HttpServer queryWebServer = new HttpServer(webManager.getBosses(), webManager.getWorkers(),
+ externalProperties.getQueryWebInterfacePort());
+ IHyracksClientConnection hcc = getHcc();
+ queryWebServer.setAttribute(HYRACKS_CONNECTION_ATTR, hcc);
+ queryWebServer.addServlet(new QueryWebInterfaceServlet(queryWebServer.ctx(), new String[] { "/*" }));
+ return queryWebServer;
+ }
+
+ protected HttpServer setupFeedServer(ExternalProperties externalProperties) throws Exception {
+ HttpServer feedServer = new HttpServer(webManager.getBosses(), webManager.getWorkers(),
+ externalProperties.getActiveServerPort());
+ feedServer.setAttribute(HYRACKS_CONNECTION_ATTR, getHcc());
+ feedServer.addServlet(new FeedServlet(feedServer.ctx(), new String[] { "/" }));
+ return feedServer;
+ }
+
+ protected IServlet createServlet(ConcurrentMap<String, Object> ctx, String key, String... paths) {
+ switch (key) {
+ case Servlets.AQL:
+ return new FullApiServlet(ctx, paths, ccExtensionManager.getAqlCompilationProvider(),
+ getStatementExecutorFactory(), componentProvider);
+ case Servlets.AQL_QUERY:
+ return new QueryApiServlet(ctx, paths, ccExtensionManager.getAqlCompilationProvider(),
+ getStatementExecutorFactory(), componentProvider);
+ case Servlets.AQL_UPDATE:
+ return new UpdateApiServlet(ctx, paths, ccExtensionManager.getAqlCompilationProvider(),
+ getStatementExecutorFactory(), componentProvider);
+ case Servlets.AQL_DDL:
+ return new DdlApiServlet(ctx, paths, ccExtensionManager.getAqlCompilationProvider(),
+ getStatementExecutorFactory(), componentProvider);
+ case Servlets.SQLPP:
+ return new FullApiServlet(ctx, paths, ccExtensionManager.getSqlppCompilationProvider(),
+ getStatementExecutorFactory(), componentProvider);
+ case Servlets.SQLPP_QUERY:
+ return new QueryApiServlet(ctx, paths, ccExtensionManager.getSqlppCompilationProvider(),
+ getStatementExecutorFactory(), componentProvider);
+ case Servlets.SQLPP_UPDATE:
+ return new UpdateApiServlet(ctx, paths, ccExtensionManager.getSqlppCompilationProvider(),
+ getStatementExecutorFactory(), componentProvider);
+ case Servlets.SQLPP_DDL:
+ return new DdlApiServlet(ctx, paths, ccExtensionManager.getSqlppCompilationProvider(),
+ getStatementExecutorFactory(), componentProvider);
+ case Servlets.QUERY_STATUS:
+ return new QueryStatusApiServlet(ctx, paths);
+ case Servlets.QUERY_RESULT:
+ return new QueryResultApiServlet(ctx, paths);
+ case Servlets.QUERY_SERVICE:
+ return new QueryServiceServlet(ctx, paths, ccExtensionManager.getSqlppCompilationProvider(),
+ getStatementExecutorFactory(), componentProvider);
+ case Servlets.CONNECTOR:
+ return new ConnectorApiServlet(ctx, paths);
+ case Servlets.SHUTDOWN:
+ return new ShutdownApiServlet(ctx, paths);
+ case Servlets.VERSION:
+ return new VersionApiServlet(ctx, paths);
+ case Servlets.CLUSTER_STATE:
+ return new ClusterApiServlet(ctx, paths);
+ case Servlets.CLUSTER_STATE_NODE_DETAIL:
+ return new NodeControllerDetailsApiServlet(ctx, paths);
+ case Servlets.CLUSTER_STATE_CC_DETAIL:
+ return new ClusterControllerDetailsApiServlet(ctx, paths);
+ case Servlets.DIAGNOSTICS:
+ return new DiagnosticsApiServlet(ctx, paths);
+ default:
+ throw new IllegalStateException(String.valueOf(key));
+ }
+ }
+
+ private IStatementExecutorFactory getStatementExecutorFactory() {
+ return ccExtensionManager.getStatementExecutorFactory(
+ ((ClusterControllerService) ccServiceCtx.getControllerService()).getExecutorService());
+ }
+
+ @Override
+ public void startupCompleted() throws Exception {
+ ClusterManagerProvider.getClusterManager().notifyStartupCompleted();
+ }
+
+ @Override
+ public IJobCapacityController getJobCapacityController() {
+ return jobCapacityController;
+ }
+
+ @Override
+ public void registerConfig(IConfigManager configManager) {
+ super.registerConfig(configManager);
+ ApplicationClassHelper.registerConfigOptions(configManager);
+ }
+
+ public static synchronized void setAsterixStateProxy(IAsterixStateProxy proxy) {
+ CCApplication.proxy = proxy;
+ }
+
+ @Override
+ public AppContextInfo getApplicationContext() {
+ return AppContextInfo.INSTANCE;
+ }
+
+ @Override
+ public IHyracksClientConnection getHcc() throws Exception {
+ String strIP = ccServiceCtx.getCCContext().getClusterControllerInfo().getClientNetAddress();
+ int port = ccServiceCtx.getCCContext().getClusterControllerInfo().getClientNetPort();
+ return new HyracksConnection(strIP, port);
+ }
+}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/953eb966/asterixdb/asterix-app/src/main/java/org/apache/asterix/hyracks/bootstrap/CCApplicationEntryPoint.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-app/src/main/java/org/apache/asterix/hyracks/bootstrap/CCApplicationEntryPoint.java b/asterixdb/asterix-app/src/main/java/org/apache/asterix/hyracks/bootstrap/CCApplicationEntryPoint.java
deleted file mode 100644
index b63bb31..0000000
--- a/asterixdb/asterix-app/src/main/java/org/apache/asterix/hyracks/bootstrap/CCApplicationEntryPoint.java
+++ /dev/null
@@ -1,316 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-package org.apache.asterix.hyracks.bootstrap;
-
-import static org.apache.asterix.api.http.servlet.ServletConstants.ASTERIX_APP_CONTEXT_INFO_ATTR;
-import static org.apache.asterix.api.http.servlet.ServletConstants.HYRACKS_CONNECTION_ATTR;
-
-import java.util.Arrays;
-import java.util.List;
-import java.util.concurrent.ConcurrentMap;
-import java.util.logging.Level;
-import java.util.logging.Logger;
-
-import org.apache.asterix.active.ActiveLifecycleListener;
-import org.apache.asterix.api.http.server.ApiServlet;
-import org.apache.asterix.api.http.server.ClusterApiServlet;
-import org.apache.asterix.api.http.server.ClusterControllerDetailsApiServlet;
-import org.apache.asterix.api.http.server.ConnectorApiServlet;
-import org.apache.asterix.api.http.server.DdlApiServlet;
-import org.apache.asterix.api.http.server.DiagnosticsApiServlet;
-import org.apache.asterix.api.http.server.FeedServlet;
-import org.apache.asterix.api.http.server.FullApiServlet;
-import org.apache.asterix.api.http.server.NodeControllerDetailsApiServlet;
-import org.apache.asterix.api.http.server.QueryApiServlet;
-import org.apache.asterix.api.http.server.QueryResultApiServlet;
-import org.apache.asterix.api.http.server.QueryServiceServlet;
-import org.apache.asterix.api.http.server.QueryStatusApiServlet;
-import org.apache.asterix.api.http.server.QueryWebInterfaceServlet;
-import org.apache.asterix.api.http.server.ShutdownApiServlet;
-import org.apache.asterix.api.http.server.UpdateApiServlet;
-import org.apache.asterix.api.http.server.VersionApiServlet;
-import org.apache.asterix.api.http.servlet.ServletConstants;
-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;
-import org.apache.asterix.common.config.AsterixExtension;
-import org.apache.asterix.common.config.ClusterProperties;
-import org.apache.asterix.common.config.ExternalProperties;
-import org.apache.asterix.common.config.MetadataProperties;
-import org.apache.asterix.common.context.IStorageComponentProvider;
-import org.apache.asterix.common.library.ILibraryManager;
-import org.apache.asterix.common.messaging.api.ICCMessageBroker;
-import org.apache.asterix.common.replication.IFaultToleranceStrategy;
-import org.apache.asterix.common.replication.IReplicationStrategy;
-import org.apache.asterix.common.utils.Servlets;
-import org.apache.asterix.external.library.ExternalLibraryManager;
-import org.apache.asterix.file.StorageComponentProvider;
-import org.apache.asterix.messaging.CCMessageBroker;
-import org.apache.asterix.metadata.MetadataManager;
-import org.apache.asterix.metadata.api.IAsterixStateProxy;
-import org.apache.asterix.metadata.bootstrap.AsterixStateProxy;
-import org.apache.asterix.metadata.cluster.ClusterManagerProvider;
-import org.apache.asterix.runtime.job.resource.JobCapacityController;
-import org.apache.asterix.runtime.utils.AppContextInfo;
-import org.apache.asterix.translator.IStatementExecutorFactory;
-import org.apache.hyracks.api.application.ICCApplicationContext;
-import org.apache.hyracks.api.client.HyracksConnection;
-import org.apache.hyracks.api.client.IHyracksClientConnection;
-import org.apache.hyracks.api.config.IConfigManager;
-import org.apache.hyracks.api.job.resource.IJobCapacityController;
-import org.apache.hyracks.api.lifecycle.LifeCycleComponentManager;
-import org.apache.hyracks.control.cc.ClusterControllerService;
-import org.apache.hyracks.control.common.controllers.CCConfig;
-import org.apache.hyracks.http.api.IServlet;
-import org.apache.hyracks.http.server.HttpServer;
-import org.apache.hyracks.http.server.WebManager;
-
-public class CCApplicationEntryPoint extends org.apache.hyracks.control.cc.CCApplicationEntryPoint {
-
- private static final Logger LOGGER = Logger.getLogger(CCApplicationEntryPoint.class.getName());
- private static IAsterixStateProxy proxy;
- protected ICCApplicationContext appCtx;
- protected CCExtensionManager ccExtensionManager;
- protected IStorageComponentProvider componentProvider;
- private IJobCapacityController jobCapacityController;
- protected WebManager webManager;
-
- @Override
- public void start(ICCApplicationContext ccAppCtx, String[] args) throws Exception {
- if (args.length > 0) {
- throw new IllegalArgumentException("Unrecognized argument(s): " + Arrays.toString(args));
- }
- final ClusterControllerService controllerService = (ClusterControllerService) ccAppCtx.getControllerService();
- ICCMessageBroker messageBroker = new CCMessageBroker(controllerService);
- this.appCtx = ccAppCtx;
-
- if (LOGGER.isLoggable(Level.INFO)) {
- LOGGER.info("Starting Asterix cluster controller");
- }
-
- appCtx.setThreadFactory(new AsterixThreadFactory(appCtx.getThreadFactory(), new LifeCycleComponentManager()));
- ILibraryManager libraryManager = new ExternalLibraryManager();
- ResourceIdManager resourceIdManager = new ResourceIdManager();
- IReplicationStrategy repStrategy = ClusterProperties.INSTANCE.getReplicationStrategy();
- IFaultToleranceStrategy ftStrategy = FaultToleranceStrategyFactory
- .create(ClusterProperties.INSTANCE.getCluster(), repStrategy, messageBroker);
- ExternalLibraryUtils.setUpExternaLibraries(libraryManager, false);
- componentProvider = new StorageComponentProvider();
- GlobalRecoveryManager.instantiate((HyracksConnection) getNewHyracksClientConnection(), componentProvider);
- AppContextInfo.initialize(appCtx, getNewHyracksClientConnection(), libraryManager, resourceIdManager,
- () -> MetadataManager.INSTANCE, GlobalRecoveryManager.instance(), ftStrategy);
- ccExtensionManager = new CCExtensionManager(getExtensions());
- AppContextInfo.INSTANCE.setExtensionManager(ccExtensionManager);
- final CCConfig ccConfig = controllerService.getCCConfig();
- if (System.getProperty("java.rmi.server.hostname") == null) {
- System.setProperty("java.rmi.server.hostname", ccConfig.getClusterListenAddress());
- }
- MetadataProperties metadataProperties = AppContextInfo.INSTANCE.getMetadataProperties();
-
- setAsterixStateProxy(AsterixStateProxy.registerRemoteObject(metadataProperties.getMetadataCallbackPort()));
- appCtx.setDistributedState(proxy);
-
- MetadataManager.initialize(proxy, metadataProperties);
-
- AppContextInfo.INSTANCE.getCCApplicationContext().addJobLifecycleListener(ActiveLifecycleListener.INSTANCE);
-
- // create event loop groups
- webManager = new WebManager();
- configureServers();
- webManager.start();
- ClusterManagerProvider.getClusterManager().registerSubscriber(GlobalRecoveryManager.instance());
- ccAppCtx.addClusterLifecycleListener(ClusterLifecycleListener.INSTANCE);
- ccAppCtx.setMessageBroker(messageBroker);
-
- jobCapacityController = new JobCapacityController(controllerService.getResourceManager());
- }
-
- protected List<AsterixExtension> getExtensions() {
- return AppContextInfo.INSTANCE.getExtensionProperties().getExtensions();
- }
-
- protected void configureServers() throws Exception {
- webManager.add(setupWebServer(AppContextInfo.INSTANCE.getExternalProperties()));
- webManager.add(setupJSONAPIServer(AppContextInfo.INSTANCE.getExternalProperties()));
- webManager.add(setupFeedServer(AppContextInfo.INSTANCE.getExternalProperties()));
- webManager.add(setupQueryWebServer(AppContextInfo.INSTANCE.getExternalProperties()));
- }
-
- @Override
- public void stop() throws Exception {
- ActiveLifecycleListener.INSTANCE.stop();
- if (LOGGER.isLoggable(Level.INFO)) {
- LOGGER.info("Stopping Asterix cluster controller");
- }
- AsterixStateProxy.unregisterRemoteObject();
- webManager.stop();
- }
-
- protected IHyracksClientConnection getNewHyracksClientConnection() throws Exception {
- String strIP = appCtx.getCCContext().getClusterControllerInfo().getClientNetAddress();
- int port = appCtx.getCCContext().getClusterControllerInfo().getClientNetPort();
- return new HyracksConnection(strIP, port);
- }
-
- protected HttpServer setupWebServer(ExternalProperties externalProperties) throws Exception {
- HttpServer webServer = new HttpServer(webManager.getBosses(), webManager.getWorkers(),
- externalProperties.getWebInterfacePort());
- IHyracksClientConnection hcc = getNewHyracksClientConnection();
- webServer.setAttribute(HYRACKS_CONNECTION_ATTR, hcc);
- webServer.addServlet(new ApiServlet(webServer.ctx(), new String[] { "/*" },
- ccExtensionManager.getAqlCompilationProvider(), ccExtensionManager.getSqlppCompilationProvider(),
- getStatementExecutorFactory(), componentProvider));
- return webServer;
- }
-
- protected HttpServer setupJSONAPIServer(ExternalProperties externalProperties) throws Exception {
- HttpServer jsonAPIServer =
- new HttpServer(webManager.getBosses(), webManager.getWorkers(), externalProperties.getAPIServerPort());
- IHyracksClientConnection hcc = getNewHyracksClientConnection();
- jsonAPIServer.setAttribute(HYRACKS_CONNECTION_ATTR, hcc);
- jsonAPIServer.setAttribute(ASTERIX_APP_CONTEXT_INFO_ATTR, AppContextInfo.INSTANCE);
- jsonAPIServer.setAttribute(ServletConstants.EXECUTOR_SERVICE,
- ((ClusterControllerService) appCtx.getControllerService()).getExecutor());
-
- // AQL rest APIs.
- addServlet(jsonAPIServer, Servlets.AQL_QUERY);
- addServlet(jsonAPIServer, Servlets.AQL_UPDATE);
- addServlet(jsonAPIServer, Servlets.AQL_DDL);
- addServlet(jsonAPIServer, Servlets.AQL);
-
- // SQL+x+ rest APIs.
- addServlet(jsonAPIServer, Servlets.SQLPP_QUERY);
- addServlet(jsonAPIServer, Servlets.SQLPP_UPDATE);
- addServlet(jsonAPIServer, Servlets.SQLPP_DDL);
- addServlet(jsonAPIServer, Servlets.SQLPP);
-
- // Other APIs.
- addServlet(jsonAPIServer, Servlets.QUERY_STATUS);
- addServlet(jsonAPIServer, Servlets.QUERY_RESULT);
- addServlet(jsonAPIServer, Servlets.QUERY_SERVICE);
- addServlet(jsonAPIServer, Servlets.CONNECTOR);
- addServlet(jsonAPIServer, Servlets.SHUTDOWN);
- addServlet(jsonAPIServer, Servlets.VERSION);
- addServlet(jsonAPIServer, Servlets.CLUSTER_STATE);
- addServlet(jsonAPIServer, Servlets.CLUSTER_STATE_NODE_DETAIL); // must not precede add of CLUSTER_STATE
- addServlet(jsonAPIServer, Servlets.CLUSTER_STATE_CC_DETAIL); // must not precede add of CLUSTER_STATE
- addServlet(jsonAPIServer, Servlets.DIAGNOSTICS);
- return jsonAPIServer;
- }
-
- protected void addServlet(HttpServer server, String path) {
- server.addServlet(createServlet(server.ctx(), path, path));
- }
-
- protected HttpServer setupQueryWebServer(ExternalProperties externalProperties) throws Exception {
- HttpServer queryWebServer = new HttpServer(webManager.getBosses(), webManager.getWorkers(),
- externalProperties.getQueryWebInterfacePort());
- IHyracksClientConnection hcc = getNewHyracksClientConnection();
- queryWebServer.setAttribute(HYRACKS_CONNECTION_ATTR, hcc);
- queryWebServer.addServlet(new QueryWebInterfaceServlet(queryWebServer.ctx(), new String[] { "/*" }));
- return queryWebServer;
- }
-
- protected HttpServer setupFeedServer(ExternalProperties externalProperties) throws Exception {
- HttpServer feedServer = new HttpServer(webManager.getBosses(), webManager.getWorkers(),
- externalProperties.getActiveServerPort());
- feedServer.setAttribute(HYRACKS_CONNECTION_ATTR, getNewHyracksClientConnection());
- feedServer.addServlet(new FeedServlet(feedServer.ctx(), new String[] { "/" }));
- return feedServer;
- }
-
- protected IServlet createServlet(ConcurrentMap<String, Object> ctx, String key, String... paths) {
- switch (key) {
- case Servlets.AQL:
- return new FullApiServlet(ctx, paths, ccExtensionManager.getAqlCompilationProvider(),
- getStatementExecutorFactory(), componentProvider);
- case Servlets.AQL_QUERY:
- return new QueryApiServlet(ctx, paths, ccExtensionManager.getAqlCompilationProvider(),
- getStatementExecutorFactory(), componentProvider);
- case Servlets.AQL_UPDATE:
- return new UpdateApiServlet(ctx, paths, ccExtensionManager.getAqlCompilationProvider(),
- getStatementExecutorFactory(), componentProvider);
- case Servlets.AQL_DDL:
- return new DdlApiServlet(ctx, paths, ccExtensionManager.getAqlCompilationProvider(),
- getStatementExecutorFactory(), componentProvider);
- case Servlets.SQLPP:
- return new FullApiServlet(ctx, paths, ccExtensionManager.getSqlppCompilationProvider(),
- getStatementExecutorFactory(), componentProvider);
- case Servlets.SQLPP_QUERY:
- return new QueryApiServlet(ctx, paths, ccExtensionManager.getSqlppCompilationProvider(),
- getStatementExecutorFactory(), componentProvider);
- case Servlets.SQLPP_UPDATE:
- return new UpdateApiServlet(ctx, paths, ccExtensionManager.getSqlppCompilationProvider(),
- getStatementExecutorFactory(), componentProvider);
- case Servlets.SQLPP_DDL:
- return new DdlApiServlet(ctx, paths, ccExtensionManager.getSqlppCompilationProvider(),
- getStatementExecutorFactory(), componentProvider);
- case Servlets.QUERY_STATUS:
- return new QueryStatusApiServlet(ctx, paths);
- case Servlets.QUERY_RESULT:
- return new QueryResultApiServlet(ctx, paths);
- case Servlets.QUERY_SERVICE:
- return new QueryServiceServlet(ctx, paths, ccExtensionManager.getSqlppCompilationProvider(),
- getStatementExecutorFactory(), componentProvider);
- case Servlets.CONNECTOR:
- return new ConnectorApiServlet(ctx, paths);
- case Servlets.SHUTDOWN:
- return new ShutdownApiServlet(ctx, paths);
- case Servlets.VERSION:
- return new VersionApiServlet(ctx, paths);
- case Servlets.CLUSTER_STATE:
- return new ClusterApiServlet(ctx, paths);
- case Servlets.CLUSTER_STATE_NODE_DETAIL:
- return new NodeControllerDetailsApiServlet(ctx, paths);
- case Servlets.CLUSTER_STATE_CC_DETAIL:
- return new ClusterControllerDetailsApiServlet(ctx, paths);
- case Servlets.DIAGNOSTICS:
- return new DiagnosticsApiServlet(ctx, paths);
- default:
- throw new IllegalStateException(String.valueOf(key));
- }
- }
-
- private IStatementExecutorFactory getStatementExecutorFactory() {
- return ccExtensionManager.getStatementExecutorFactory(
- ((ClusterControllerService) appCtx.getControllerService()).getExecutorService());
- }
-
- @Override
- public void startupCompleted() throws Exception {
- ClusterManagerProvider.getClusterManager().notifyStartupCompleted();
- }
-
- @Override
- public IJobCapacityController getJobCapacityController() {
- return jobCapacityController;
- }
-
- @Override
- public void registerConfig(IConfigManager configManager) {
- super.registerConfig(configManager);
- ApplicationEntryPointHelper.registerConfigOptions(configManager);
- }
-
- public static synchronized void setAsterixStateProxy(IAsterixStateProxy proxy) {
- org.apache.asterix.hyracks.bootstrap.CCApplicationEntryPoint.proxy = proxy;
- }
-}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/953eb966/asterixdb/asterix-app/src/main/java/org/apache/asterix/hyracks/bootstrap/NCApplication.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-app/src/main/java/org/apache/asterix/hyracks/bootstrap/NCApplication.java b/asterixdb/asterix-app/src/main/java/org/apache/asterix/hyracks/bootstrap/NCApplication.java
new file mode 100644
index 0000000..28e1f23
--- /dev/null
+++ b/asterixdb/asterix-app/src/main/java/org/apache/asterix/hyracks/bootstrap/NCApplication.java
@@ -0,0 +1,254 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.asterix.hyracks.bootstrap;
+
+import java.io.File;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.logging.Level;
+import java.util.logging.Logger;
+
+import org.apache.asterix.app.nc.NCAppRuntimeContext;
+import org.apache.asterix.app.replication.message.StartupTaskRequestMessage;
+import org.apache.asterix.common.api.AsterixThreadFactory;
+import org.apache.asterix.common.api.IAppRuntimeContext;
+import org.apache.asterix.common.config.AsterixExtension;
+import org.apache.asterix.common.config.ClusterProperties;
+import org.apache.asterix.common.config.IPropertiesProvider;
+import org.apache.asterix.common.config.MessagingProperties;
+import org.apache.asterix.common.config.MetadataProperties;
+import org.apache.asterix.common.config.NodeProperties;
+import org.apache.asterix.common.config.StorageProperties;
+import org.apache.asterix.common.config.TransactionProperties;
+import org.apache.asterix.common.transactions.IRecoveryManager;
+import org.apache.asterix.common.transactions.IRecoveryManager.SystemState;
+import org.apache.asterix.common.utils.PrintUtil;
+import org.apache.asterix.common.utils.StoragePathUtil;
+import org.apache.asterix.event.schema.cluster.Cluster;
+import org.apache.asterix.event.schema.cluster.Node;
+import org.apache.asterix.messaging.MessagingChannelInterfaceFactory;
+import org.apache.asterix.messaging.NCMessageBroker;
+import org.apache.asterix.transaction.management.resource.PersistentLocalResourceRepository;
+import org.apache.commons.io.FileUtils;
+import org.apache.hyracks.api.application.INCServiceContext;
+import org.apache.hyracks.api.application.IServiceContext;
+import org.apache.hyracks.api.config.IConfigManager;
+import org.apache.hyracks.api.job.resource.NodeCapacity;
+import org.apache.hyracks.api.messages.IMessageBroker;
+import org.apache.hyracks.control.common.controllers.NCConfig;
+import org.apache.hyracks.control.nc.BaseNCApplication;
+import org.apache.hyracks.control.nc.NodeControllerService;
+
+public class NCApplication extends BaseNCApplication {
+ private static final Logger LOGGER = Logger.getLogger(NCApplication.class.getName());
+
+ private INCServiceContext ncServiceCtx;
+ private IAppRuntimeContext runtimeContext;
+ private String nodeId;
+ private boolean stopInitiated = false;
+ private SystemState systemState;
+
+ @Override
+ public void registerConfig(IConfigManager configManager) {
+ super.registerConfig(configManager);
+ ApplicationClassHelper.registerConfigOptions(configManager);
+ }
+
+ @Override
+ public void start(IServiceContext serviceCtx, String[] args) throws Exception {
+ if (args.length > 0) {
+ throw new IllegalArgumentException("Unrecognized argument(s): " + Arrays.toString(args));
+ }
+ this.ncServiceCtx = (INCServiceContext) serviceCtx;
+ ncServiceCtx.setThreadFactory(
+ new AsterixThreadFactory(ncServiceCtx.getThreadFactory(), ncServiceCtx.getLifeCycleComponentManager()));
+ nodeId = this.ncServiceCtx.getNodeId();
+ if (LOGGER.isLoggable(Level.INFO)) {
+ LOGGER.info("Starting Asterix node controller: " + nodeId);
+ }
+
+ final NodeControllerService controllerService = (NodeControllerService) ncServiceCtx.getControllerService();
+
+ if (System.getProperty("java.rmi.server.hostname") == null) {
+ System.setProperty("java.rmi.server.hostname",
+ (controllerService).getConfiguration().getClusterPublicAddress());
+ }
+ runtimeContext = new NCAppRuntimeContext(this.ncServiceCtx, getExtensions());
+ MetadataProperties metadataProperties = runtimeContext.getMetadataProperties();
+ if (!metadataProperties.getNodeNames().contains(this.ncServiceCtx.getNodeId())) {
+ if (LOGGER.isLoggable(Level.INFO)) {
+ LOGGER.info("Substitute node joining : " + this.ncServiceCtx.getNodeId());
+ }
+ updateOnNodeJoin();
+ }
+ runtimeContext.initialize(runtimeContext.getNodeProperties().isInitialRun());
+ MessagingProperties messagingProperties = runtimeContext.getMessagingProperties();
+ IMessageBroker messageBroker = new NCMessageBroker(controllerService, messagingProperties);
+ this.ncServiceCtx.setMessageBroker(messageBroker);
+ MessagingChannelInterfaceFactory interfaceFactory =
+ new MessagingChannelInterfaceFactory((NCMessageBroker) messageBroker, messagingProperties);
+ this.ncServiceCtx.setMessagingChannelInterfaceFactory(interfaceFactory);
+
+ IRecoveryManager recoveryMgr = runtimeContext.getTransactionSubsystem().getRecoveryManager();
+ systemState = recoveryMgr.getSystemState();
+
+ if (systemState == SystemState.PERMANENT_DATA_LOSS) {
+ if (LOGGER.isLoggable(Level.INFO)) {
+ LOGGER.info("System state: " + SystemState.PERMANENT_DATA_LOSS);
+ LOGGER.info("Node ID: " + nodeId);
+ LOGGER.info("Stores: " + PrintUtil.toString(metadataProperties.getStores()));
+ LOGGER.info("Root Metadata Store: " + metadataProperties.getStores().get(nodeId)[0]);
+ }
+ PersistentLocalResourceRepository localResourceRepository =
+ (PersistentLocalResourceRepository) runtimeContext.getLocalResourceRepository();
+ localResourceRepository.initializeNewUniverse(ClusterProperties.INSTANCE.getStorageDirectoryName());
+ }
+
+ performLocalCleanUp();
+ }
+
+ protected List<AsterixExtension> getExtensions() {
+ return Collections.emptyList();
+ }
+
+ @Override
+ public void stop() throws Exception {
+ if (!stopInitiated) {
+ runtimeContext.setShuttingdown(true);
+ stopInitiated = true;
+ if (LOGGER.isLoggable(Level.INFO)) {
+ LOGGER.info("Stopping Asterix node controller: " + nodeId);
+ }
+ //Clean any temporary files
+ performLocalCleanUp();
+
+ //Note: stopping recovery manager will make a sharp checkpoint
+ ncServiceCtx.getLifeCycleComponentManager().stopAll(false);
+ runtimeContext.deinitialize();
+ } else {
+ if (LOGGER.isLoggable(Level.INFO)) {
+ LOGGER.info("Duplicate attempt to stop ignored: " + nodeId);
+ }
+ }
+ }
+
+ @Override
+ public void startupCompleted() throws Exception {
+ // Since we don't pass initial run flag in AsterixHyracksIntegrationUtil, we use the virtualNC flag
+ final NodeProperties nodeProperties = runtimeContext.getNodeProperties();
+ if (systemState == SystemState.PERMANENT_DATA_LOSS
+ && (nodeProperties.isInitialRun() || nodeProperties.isVirtualNc())) {
+ systemState = SystemState.BOOTSTRAPPING;
+ }
+ // Request startup tasks from CC
+ StartupTaskRequestMessage.send((NodeControllerService) ncServiceCtx.getControllerService(), systemState);
+ }
+
+ @Override
+ public NodeCapacity getCapacity() {
+ IPropertiesProvider propertiesProvider = runtimeContext;
+ StorageProperties storageProperties = propertiesProvider.getStorageProperties();
+ // Deducts the reserved buffer cache size and memory component size from the maxium heap size,
+ // and deducts one core for processing heartbeats.
+ long memorySize = Runtime.getRuntime().maxMemory() - storageProperties.getBufferCacheSize()
+ - storageProperties.getMemoryComponentGlobalBudget();
+ int allCores = Runtime.getRuntime().availableProcessors();
+ int maximumCoresForComputation = allCores > 1 ? allCores - 1 : allCores;
+ return new NodeCapacity(memorySize, maximumCoresForComputation);
+ }
+
+ private void performLocalCleanUp() {
+ //Delete working area files from failed jobs
+ runtimeContext.getIOManager().deleteWorkspaceFiles();
+
+ //Reclaim storage for temporary datasets.
+ String storageDirName = ClusterProperties.INSTANCE.getStorageDirectoryName();
+ String[] ioDevices = ((PersistentLocalResourceRepository) runtimeContext.getLocalResourceRepository())
+ .getStorageMountingPoints();
+ for (String ioDevice : ioDevices) {
+ String tempDatasetsDir =
+ ioDevice + storageDirName + File.separator + StoragePathUtil.TEMP_DATASETS_STORAGE_FOLDER;
+ FileUtils.deleteQuietly(new File(tempDatasetsDir));
+ }
+
+ //TODO
+ //Reclaim storage for orphaned index artifacts in NCs.
+ //Note: currently LSM indexes invalid components are deleted when an index is activated.
+ }
+
+ private void updateOnNodeJoin() {
+ MetadataProperties metadataProperties = runtimeContext.getMetadataProperties();
+ if (!metadataProperties.getNodeNames().contains(nodeId)) {
+ Cluster cluster = ClusterProperties.INSTANCE.getCluster();
+ if (cluster == null) {
+ throw new IllegalStateException("No cluster configuration found for this instance");
+ }
+ NCConfig ncConfig = ((NodeControllerService) ncServiceCtx.getControllerService()).getConfiguration();
+ ncConfig.getConfigManager().registerVirtualNode(nodeId);
+ String asterixInstanceName = metadataProperties.getInstanceName();
+ TransactionProperties txnProperties = runtimeContext.getTransactionProperties();
+ Node self = null;
+ List<Node> nodes;
+ if (cluster.getSubstituteNodes() != null) {
+ nodes = cluster.getSubstituteNodes().getNode();
+ } else {
+ throw new IllegalStateException("Unknown node joining the cluster");
+ }
+ for (Node node : nodes) {
+ String ncId = asterixInstanceName + "_" + node.getId();
+ if (ncId.equalsIgnoreCase(nodeId)) {
+ String storeDir = ClusterProperties.INSTANCE.getStorageDirectoryName();
+ String nodeIoDevices = node.getIodevices() == null ? cluster.getIodevices() : node.getIodevices();
+ String[] ioDevicePaths = nodeIoDevices.trim().split(",");
+ for (int i = 0; i < ioDevicePaths.length; i++) {
+ // construct full store path
+ ioDevicePaths[i] += File.separator + storeDir;
+ }
+ metadataProperties.getStores().put(nodeId, ioDevicePaths);
+
+ String coredumpPath = node.getLogDir() == null ? cluster.getLogDir() : node.getLogDir();
+ metadataProperties.getCoredumpPaths().put(nodeId, coredumpPath);
+
+ String txnLogDir = node.getTxnLogDir() == null ? cluster.getTxnLogDir() : node.getTxnLogDir();
+ txnProperties.getLogDirectories().put(nodeId, txnLogDir);
+
+ if (LOGGER.isLoggable(Level.INFO)) {
+ LOGGER.info("Store set to : " + storeDir);
+ LOGGER.info("Coredump dir set to : " + coredumpPath);
+ LOGGER.info("Transaction log dir set to :" + txnLogDir);
+ }
+ self = node;
+ break;
+ }
+ }
+ if (self != null) {
+ cluster.getSubstituteNodes().getNode().remove(self);
+ cluster.getNode().add(self);
+ } else {
+ throw new IllegalStateException("Unknown node joining the cluster");
+ }
+ }
+ }
+
+ @Override
+ public IAppRuntimeContext getApplicationContext() {
+ return runtimeContext;
+ }
+}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/953eb966/asterixdb/asterix-app/src/main/java/org/apache/asterix/hyracks/bootstrap/NCApplicationEntryPoint.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-app/src/main/java/org/apache/asterix/hyracks/bootstrap/NCApplicationEntryPoint.java b/asterixdb/asterix-app/src/main/java/org/apache/asterix/hyracks/bootstrap/NCApplicationEntryPoint.java
deleted file mode 100644
index 07a9a61..0000000
--- a/asterixdb/asterix-app/src/main/java/org/apache/asterix/hyracks/bootstrap/NCApplicationEntryPoint.java
+++ /dev/null
@@ -1,248 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-package org.apache.asterix.hyracks.bootstrap;
-
-import java.io.File;
-import java.util.Arrays;
-import java.util.Collections;
-import java.util.List;
-import java.util.logging.Level;
-import java.util.logging.Logger;
-
-import org.apache.asterix.app.nc.NCAppRuntimeContext;
-import org.apache.asterix.app.replication.message.StartupTaskRequestMessage;
-import org.apache.asterix.common.api.AsterixThreadFactory;
-import org.apache.asterix.common.api.IAppRuntimeContext;
-import org.apache.asterix.common.config.AsterixExtension;
-import org.apache.asterix.common.config.ClusterProperties;
-import org.apache.asterix.common.config.IPropertiesProvider;
-import org.apache.asterix.common.config.MessagingProperties;
-import org.apache.asterix.common.config.MetadataProperties;
-import org.apache.asterix.common.config.NodeProperties;
-import org.apache.asterix.common.config.StorageProperties;
-import org.apache.asterix.common.config.TransactionProperties;
-import org.apache.asterix.common.transactions.IRecoveryManager;
-import org.apache.asterix.common.transactions.IRecoveryManager.SystemState;
-import org.apache.asterix.common.utils.PrintUtil;
-import org.apache.asterix.common.utils.StoragePathUtil;
-import org.apache.asterix.event.schema.cluster.Cluster;
-import org.apache.asterix.event.schema.cluster.Node;
-import org.apache.asterix.messaging.MessagingChannelInterfaceFactory;
-import org.apache.asterix.messaging.NCMessageBroker;
-import org.apache.asterix.transaction.management.resource.PersistentLocalResourceRepository;
-import org.apache.commons.io.FileUtils;
-import org.apache.hyracks.api.application.INCApplicationContext;
-import org.apache.hyracks.api.config.IConfigManager;
-import org.apache.hyracks.api.job.resource.NodeCapacity;
-import org.apache.hyracks.api.messages.IMessageBroker;
-import org.apache.hyracks.control.common.controllers.NCConfig;
-import org.apache.hyracks.control.nc.NodeControllerService;
-
-public class NCApplicationEntryPoint extends org.apache.hyracks.control.nc.NCApplicationEntryPoint {
- private static final Logger LOGGER = Logger.getLogger(NCApplicationEntryPoint.class.getName());
-
- private INCApplicationContext ncAppCtx;
- private IAppRuntimeContext runtimeContext;
- private String nodeId;
- private boolean stopInitiated = false;
- private SystemState systemState;
-
- @Override
- public void registerConfigOptions(IConfigManager configManager) {
- super.registerConfigOptions(configManager);
- ApplicationEntryPointHelper.registerConfigOptions(configManager);
- }
-
- @Override
- public void start(INCApplicationContext ncAppCtx, String[] args) throws Exception {
- if (args.length > 0) {
- throw new IllegalArgumentException("Unrecognized argument(s): " + Arrays.toString(args));
- }
- ncAppCtx.setThreadFactory(new AsterixThreadFactory(ncAppCtx.getThreadFactory(),
- ncAppCtx.getLifeCycleComponentManager()));
- this.ncAppCtx = ncAppCtx;
- nodeId = this.ncAppCtx.getNodeId();
- if (LOGGER.isLoggable(Level.INFO)) {
- LOGGER.info("Starting Asterix node controller: " + nodeId);
- }
-
- final NodeControllerService controllerService = (NodeControllerService) ncAppCtx.getControllerService();
-
- if (System.getProperty("java.rmi.server.hostname") == null) {
- System.setProperty("java.rmi.server.hostname", (controllerService)
- .getConfiguration().getClusterPublicAddress());
- }
- runtimeContext = new NCAppRuntimeContext(this.ncAppCtx, getExtensions());
- MetadataProperties metadataProperties = runtimeContext.getMetadataProperties();
- if (!metadataProperties.getNodeNames().contains(this.ncAppCtx.getNodeId())) {
- if (LOGGER.isLoggable(Level.INFO)) {
- LOGGER.info("Substitute node joining : " + this.ncAppCtx.getNodeId());
- }
- updateOnNodeJoin();
- }
- runtimeContext.initialize(runtimeContext.getNodeProperties().isInitialRun());
- this.ncAppCtx.setApplicationObject(runtimeContext);
- MessagingProperties messagingProperties = runtimeContext.getMessagingProperties();
- IMessageBroker messageBroker = new NCMessageBroker(controllerService, messagingProperties);
- this.ncAppCtx.setMessageBroker(messageBroker);
- MessagingChannelInterfaceFactory interfaceFactory = new MessagingChannelInterfaceFactory(
- (NCMessageBroker) messageBroker, messagingProperties);
- this.ncAppCtx.setMessagingChannelInterfaceFactory(interfaceFactory);
-
- IRecoveryManager recoveryMgr = runtimeContext.getTransactionSubsystem().getRecoveryManager();
- systemState = recoveryMgr.getSystemState();
-
- if (systemState == SystemState.PERMANENT_DATA_LOSS) {
- if (LOGGER.isLoggable(Level.INFO)) {
- LOGGER.info("System state: " + SystemState.PERMANENT_DATA_LOSS);
- LOGGER.info("Node ID: " + nodeId);
- LOGGER.info("Stores: " + PrintUtil.toString(metadataProperties.getStores()));
- LOGGER.info("Root Metadata Store: " + metadataProperties.getStores().get(nodeId)[0]);
- }
- PersistentLocalResourceRepository localResourceRepository =
- (PersistentLocalResourceRepository) runtimeContext.getLocalResourceRepository();
- localResourceRepository.initializeNewUniverse(ClusterProperties.INSTANCE.getStorageDirectoryName());
- }
-
- performLocalCleanUp();
- }
-
- protected List<AsterixExtension> getExtensions() {
- return Collections.emptyList();
- }
-
- @Override
- public void stop() throws Exception {
- if (!stopInitiated) {
- runtimeContext.setShuttingdown(true);
- stopInitiated = true;
- if (LOGGER.isLoggable(Level.INFO)) {
- LOGGER.info("Stopping Asterix node controller: " + nodeId);
- }
- //Clean any temporary files
- performLocalCleanUp();
-
- //Note: stopping recovery manager will make a sharp checkpoint
- ncAppCtx.getLifeCycleComponentManager().stopAll(false);
- runtimeContext.deinitialize();
- } else {
- if (LOGGER.isLoggable(Level.INFO)) {
- LOGGER.info("Duplicate attempt to stop ignored: " + nodeId);
- }
- }
- }
-
- @Override
- public void notifyStartupComplete() throws Exception {
- // Since we don't pass initial run flag in AsterixHyracksIntegrationUtil, we use the virtualNC flag
- final NodeProperties nodeProperties = runtimeContext.getNodeProperties();
- if (systemState == SystemState.PERMANENT_DATA_LOSS && (nodeProperties.isInitialRun() || nodeProperties.isVirtualNc())) {
- systemState = SystemState.BOOTSTRAPPING;
- }
- // Request startup tasks from CC
- StartupTaskRequestMessage.send((NodeControllerService) ncAppCtx.getControllerService(),
- systemState);
- }
-
- @Override
- public NodeCapacity getCapacity() {
- IPropertiesProvider propertiesProvider = runtimeContext;
- StorageProperties storageProperties = propertiesProvider.getStorageProperties();
- // Deducts the reserved buffer cache size and memory component size from the maxium heap size,
- // and deducts one core for processing heartbeats.
- long memorySize = Runtime.getRuntime().maxMemory() - storageProperties.getBufferCacheSize()
- - storageProperties.getMemoryComponentGlobalBudget();
- int allCores = Runtime.getRuntime().availableProcessors();
- int maximumCoresForComputation = allCores > 1 ? allCores - 1 : allCores;
- return new NodeCapacity(memorySize, maximumCoresForComputation);
- }
-
- private void performLocalCleanUp() {
- //Delete working area files from failed jobs
- runtimeContext.getIOManager().deleteWorkspaceFiles();
-
- //Reclaim storage for temporary datasets.
- String storageDirName = ClusterProperties.INSTANCE.getStorageDirectoryName();
- String[] ioDevices = ((PersistentLocalResourceRepository) runtimeContext.getLocalResourceRepository())
- .getStorageMountingPoints();
- for (String ioDevice : ioDevices) {
- String tempDatasetsDir = ioDevice + storageDirName + File.separator
- + StoragePathUtil.TEMP_DATASETS_STORAGE_FOLDER;
- FileUtils.deleteQuietly(new File(tempDatasetsDir));
- }
-
- //TODO
- //Reclaim storage for orphaned index artifacts in NCs.
- //Note: currently LSM indexes invalid components are deleted when an index is activated.
- }
-
- private void updateOnNodeJoin() {
- MetadataProperties metadataProperties = runtimeContext.getMetadataProperties();
- if (!metadataProperties.getNodeNames().contains(nodeId)) {
- Cluster cluster = ClusterProperties.INSTANCE.getCluster();
- if (cluster == null) {
- throw new IllegalStateException("No cluster configuration found for this instance");
- }
- NCConfig ncConfig = ((NodeControllerService) ncAppCtx.getControllerService()).getConfiguration();
- ncConfig.getConfigManager().registerVirtualNode(nodeId);
- String asterixInstanceName = metadataProperties.getInstanceName();
- TransactionProperties txnProperties = runtimeContext.getTransactionProperties();
- Node self = null;
- List<Node> nodes;
- if (cluster.getSubstituteNodes() != null) {
- nodes = cluster.getSubstituteNodes().getNode();
- } else {
- throw new IllegalStateException("Unknown node joining the cluster");
- }
- for (Node node : nodes) {
- String ncId = asterixInstanceName + "_" + node.getId();
- if (ncId.equalsIgnoreCase(nodeId)) {
- String storeDir = ClusterProperties.INSTANCE.getStorageDirectoryName();
- String nodeIoDevices = node.getIodevices() == null ? cluster.getIodevices() : node.getIodevices();
- String[] ioDevicePaths = nodeIoDevices.trim().split(",");
- for (int i = 0; i < ioDevicePaths.length; i++) {
- // construct full store path
- ioDevicePaths[i] += File.separator + storeDir;
- }
- metadataProperties.getStores().put(nodeId, ioDevicePaths);
-
- String coredumpPath = node.getLogDir() == null ? cluster.getLogDir() : node.getLogDir();
- metadataProperties.getCoredumpPaths().put(nodeId, coredumpPath);
-
- String txnLogDir = node.getTxnLogDir() == null ? cluster.getTxnLogDir() : node.getTxnLogDir();
- txnProperties.getLogDirectories().put(nodeId, txnLogDir);
-
- if (LOGGER.isLoggable(Level.INFO)) {
- LOGGER.info("Store set to : " + storeDir);
- LOGGER.info("Coredump dir set to : " + coredumpPath);
- LOGGER.info("Transaction log dir set to :" + txnLogDir);
- }
- self = node;
- break;
- }
- }
- if (self != null) {
- cluster.getSubstituteNodes().getNode().remove(self);
- cluster.getNode().add(self);
- } else {
- throw new IllegalStateException("Unknown node joining the cluster");
- }
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/953eb966/asterixdb/asterix-app/src/main/java/org/apache/asterix/messaging/NCMessageBroker.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-app/src/main/java/org/apache/asterix/messaging/NCMessageBroker.java b/asterixdb/asterix-app/src/main/java/org/apache/asterix/messaging/NCMessageBroker.java
index 10193b3..f615138 100644
--- a/asterixdb/asterix-app/src/main/java/org/apache/asterix/messaging/NCMessageBroker.java
+++ b/asterixdb/asterix-app/src/main/java/org/apache/asterix/messaging/NCMessageBroker.java
@@ -46,7 +46,7 @@ public class NCMessageBroker implements INCMessageBroker {
public NCMessageBroker(NodeControllerService ncs, MessagingProperties messagingProperties) {
this.ncs = ncs;
- appContext = (IAppRuntimeContext) ncs.getApplicationContext().getApplicationObject();
+ appContext = (IAppRuntimeContext) ncs.getApplicationContext();
maxMsgSize = messagingProperties.getFrameSize();
int messagingMemoryBudget = messagingProperties.getFrameSize() * messagingProperties.getFrameCount();
messagingFramePool = new ConcurrentFramePool(ncs.getId(), messagingMemoryBudget,
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/953eb966/asterixdb/asterix-app/src/main/java/org/apache/asterix/utils/FeedOperations.java
----------------------------------------------------------------------
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 f8b5496..4ea524a 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
@@ -391,7 +391,7 @@ public class FeedOperations {
private static void SendActiveMessage(ActiveManagerMessage activeManagerMessage, String nodeId) throws Exception {
ICCMessageBroker messageBroker =
- (ICCMessageBroker) AppContextInfo.INSTANCE.getCCApplicationContext().getMessageBroker();
+ (ICCMessageBroker) AppContextInfo.INSTANCE.getCCServiceContext().getMessageBroker();
messageBroker.sendApplicationMessageToNC(activeManagerMessage, nodeId);
}
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/953eb966/asterixdb/asterix-app/src/test/java/org/apache/asterix/app/bootstrap/TestNodeController.java
----------------------------------------------------------------------
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 2061cda..b7026ef 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
@@ -141,8 +141,7 @@ public class TestNodeController {
throw th;
}
jobletCtx = Mockito.mock(IHyracksJobletContext.class);
- Mockito.when(jobletCtx.getApplicationContext())
- .thenReturn(ExecutionTestUtil.integrationUtil.ncs[0].getApplicationContext());
+ Mockito.when(jobletCtx.getServiceContext()).thenReturn(ExecutionTestUtil.integrationUtil.ncs[0].getContext());
Mockito.when(jobletCtx.getJobId()).thenAnswer(new Answer<JobId>() {
@Override
public JobId answer(InvocationOnMock invocation) throws Throwable {
@@ -389,7 +388,7 @@ public class TestNodeController {
public TransactionSubsystem getTransactionSubsystem() {
return (TransactionSubsystem) ((NCAppRuntimeContext) ExecutionTestUtil.integrationUtil.ncs[0]
- .getApplicationContext().getApplicationObject()).getTransactionSubsystem();
+ .getApplicationContext()).getTransactionSubsystem();
}
public ITransactionManager getTransactionManager() {
@@ -397,8 +396,7 @@ public class TestNodeController {
}
public NCAppRuntimeContext getAppRuntimeContext() {
- return (NCAppRuntimeContext) ExecutionTestUtil.integrationUtil.ncs[0].getApplicationContext()
- .getApplicationObject();
+ return (NCAppRuntimeContext) ExecutionTestUtil.integrationUtil.ncs[0].getApplicationContext();
}
public DatasetLifecycleManager getDatasetLifecycleManager() {
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/953eb966/asterixdb/asterix-app/src/test/java/org/apache/asterix/common/config/ConfigUsageTest.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-app/src/test/java/org/apache/asterix/common/config/ConfigUsageTest.java b/asterixdb/asterix-app/src/test/java/org/apache/asterix/common/config/ConfigUsageTest.java
index 9cd961a..6a8dabf 100644
--- a/asterixdb/asterix-app/src/test/java/org/apache/asterix/common/config/ConfigUsageTest.java
+++ b/asterixdb/asterix-app/src/test/java/org/apache/asterix/common/config/ConfigUsageTest.java
@@ -31,7 +31,7 @@ import java.util.TreeSet;
import java.util.function.Predicate;
import java.util.stream.Collectors;
-import org.apache.asterix.hyracks.bootstrap.CCApplicationEntryPoint;
+import org.apache.asterix.hyracks.bootstrap.CCApplication;
import org.apache.commons.lang3.StringEscapeUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.hyracks.api.config.IOption;
@@ -83,8 +83,8 @@ public class ConfigUsageTest {
protected ConfigManager getConfigManager() {
ConfigManager configManager = new ConfigManager();
- CCApplicationEntryPoint aep = new CCApplicationEntryPoint();
- aep.registerConfig(configManager);
+ CCApplication application = new CCApplication();
+ application.registerConfig(configManager);
ControllerConfig.defaultDir = ControllerConfig.defaultDir.replace(System.getProperty("java.io.tmpdir"),
"${java.io.tmpdir}/");
return configManager;
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/953eb966/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/runtime/ExecutionTestUtil.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/runtime/ExecutionTestUtil.java b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/runtime/ExecutionTestUtil.java
index 90eb441..8714319 100644
--- a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/runtime/ExecutionTestUtil.java
+++ b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/runtime/ExecutionTestUtil.java
@@ -92,8 +92,7 @@ public class ExecutionTestUtil {
libraryManagers.add(AppContextInfo.INSTANCE.getLibraryManager());
// Adds library managers for NCs, one-per-NC.
for (NodeControllerService nc : integrationUtil.ncs) {
- IAppRuntimeContext runtimeCtx = (IAppRuntimeContext) nc.getApplicationContext()
- .getApplicationObject();
+ IAppRuntimeContext runtimeCtx = (IAppRuntimeContext) nc.getApplicationContext();
libraryManagers.add(runtimeCtx.getLibraryManager());
}
return libraryManagers;
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/953eb966/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/context/AsterixVirtualBufferCacheProvider.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/context/AsterixVirtualBufferCacheProvider.java b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/context/AsterixVirtualBufferCacheProvider.java
index af972b6..7f3f6aa 100644
--- a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/context/AsterixVirtualBufferCacheProvider.java
+++ b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/context/AsterixVirtualBufferCacheProvider.java
@@ -56,7 +56,7 @@ public class AsterixVirtualBufferCacheProvider implements IVirtualBufferCachePro
deviceId = i;
}
}
- return ((IAppRuntimeContext) ctx.getJobletContext().getApplicationContext().getApplicationObject())
+ return ((IAppRuntimeContext) ctx.getJobletContext().getServiceContext().getApplicationContext())
.getDatasetLifecycleManager().getVirtualBufferCaches(datasetID, deviceId);
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/953eb966/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/context/CorrelatedPrefixMergePolicyFactory.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/context/CorrelatedPrefixMergePolicyFactory.java b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/context/CorrelatedPrefixMergePolicyFactory.java
index b2bcf17..d9381a2 100644
--- a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/context/CorrelatedPrefixMergePolicyFactory.java
+++ b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/context/CorrelatedPrefixMergePolicyFactory.java
@@ -37,14 +37,14 @@ public class CorrelatedPrefixMergePolicyFactory implements ILSMMergePolicyFactor
private static final String[] SET_VALUES = new String[] { "max-mergable-component-size",
"max-tolerance-component-count" };
- private static final Set<String> PROPERTIES_NAMES = new HashSet<String>(Arrays.asList(SET_VALUES));
+ private static final Set<String> PROPERTIES_NAMES = new HashSet<>(Arrays.asList(SET_VALUES));
private int datasetID;
@Override
public ILSMMergePolicy createMergePolicy(Map<String, String> properties, IHyracksTaskContext ctx) {
IDatasetLifecycleManager dslcManager = ((IAppRuntimeContext) ctx.getJobletContext()
- .getApplicationContext().getApplicationObject()).getDatasetLifecycleManager();
+ .getServiceContext().getApplicationContext()).getDatasetLifecycleManager();
ILSMMergePolicy policy = new CorrelatedPrefixMergePolicy(dslcManager, datasetID);
policy.configure(properties);
return policy;
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/953eb966/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/context/TransactionSubsystemProvider.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/context/TransactionSubsystemProvider.java b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/context/TransactionSubsystemProvider.java
index cfbad43..5dab970 100644
--- a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/context/TransactionSubsystemProvider.java
+++ b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/context/TransactionSubsystemProvider.java
@@ -37,8 +37,8 @@ public class TransactionSubsystemProvider implements ITransactionSubsystemProvid
@Override
public ITransactionSubsystem getTransactionSubsystem(IHyracksTaskContext ctx) {
- IAppRuntimeContext runtimeCtx = (IAppRuntimeContext) ctx.getJobletContext()
- .getApplicationContext().getApplicationObject();
- return runtimeCtx.getTransactionSubsystem();
+ IAppRuntimeContext appCtx =
+ (IAppRuntimeContext) ctx.getJobletContext().getServiceContext().getApplicationContext();
+ return appCtx.getTransactionSubsystem();
}
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/953eb966/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/dataflow/IApplicationContextInfo.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/dataflow/IApplicationContextInfo.java b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/dataflow/IApplicationContextInfo.java
index eb944a2..3c5328d 100644
--- a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/dataflow/IApplicationContextInfo.java
+++ b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/dataflow/IApplicationContextInfo.java
@@ -20,7 +20,7 @@ package org.apache.asterix.common.dataflow;
import org.apache.asterix.common.cluster.IGlobalRecoveryManager;
import org.apache.asterix.common.library.ILibraryManager;
-import org.apache.hyracks.api.application.ICCApplicationContext;
+import org.apache.hyracks.api.application.ICCServiceContext;
import org.apache.hyracks.storage.am.common.api.IIndexLifecycleManagerProvider;
import org.apache.hyracks.storage.common.IStorageManager;
@@ -28,7 +28,7 @@ import org.apache.hyracks.storage.common.IStorageManager;
* Provides methods for obtaining
* {@link org.apache.hyracks.storage.am.common.api.IIndexLifecycleManagerProvider},
* {@link org.apache.hyracks.storage.common.IStorageManager},
- * {@link org.apache.hyracks.api.application.ICCApplicationContext},
+ * {@link org.apache.hyracks.api.application.ICCServiceContext},
* {@link org.apache.asterix.common.cluster.IGlobalRecoveryManager},
* and {@link org.apache.asterix.common.library.ILibraryManager}
* at the cluster controller side.
@@ -48,9 +48,9 @@ public interface IApplicationContextInfo {
public IStorageManager getStorageManager();
/**
- * @return an instance which implements {@link org.apache.hyracks.api.application.ICCApplicationContext}
+ * @return an instance which implements {@link org.apache.hyracks.api.application.ICCServiceContext}
*/
- public ICCApplicationContext getCCApplicationContext();
+ public ICCServiceContext getCCServiceContext();
/**
* @return the global recovery manager which implements
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/953eb966/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/dataflow/LSMInsertDeleteOperatorNodePushable.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/dataflow/LSMInsertDeleteOperatorNodePushable.java b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/dataflow/LSMInsertDeleteOperatorNodePushable.java
index ac878ec..2eca55d 100644
--- a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/dataflow/LSMInsertDeleteOperatorNodePushable.java
+++ b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/dataflow/LSMInsertDeleteOperatorNodePushable.java
@@ -21,13 +21,13 @@ package org.apache.asterix.common.dataflow;
import java.nio.ByteBuffer;
import org.apache.asterix.common.api.IAppRuntimeContext;
-import org.apache.hyracks.api.exceptions.ErrorCode;
import org.apache.asterix.common.transactions.ILogMarkerCallback;
import org.apache.asterix.common.transactions.PrimaryIndexLogMarkerCallback;
import org.apache.hyracks.api.comm.VSizeFrame;
import org.apache.hyracks.api.context.IHyracksTaskContext;
import org.apache.hyracks.api.dataflow.value.IRecordDescriptorProvider;
import org.apache.hyracks.api.dataflow.value.RecordDescriptor;
+import org.apache.hyracks.api.exceptions.ErrorCode;
import org.apache.hyracks.api.exceptions.HyracksDataException;
import org.apache.hyracks.dataflow.common.comm.io.FrameTupleAccessor;
import org.apache.hyracks.dataflow.common.comm.io.FrameTupleAppender;
@@ -94,7 +94,7 @@ public class LSMInsertDeleteOperatorNodePushable extends LSMIndexInsertUpdateDel
frameTuple = new FrameTupleReference();
}
IAppRuntimeContext runtimeCtx =
- (IAppRuntimeContext) ctx.getJobletContext().getApplicationContext().getApplicationObject();
+ (IAppRuntimeContext) ctx.getJobletContext().getServiceContext().getApplicationContext();
LSMIndexUtil.checkAndSetFirstLSN(lsmIndex, runtimeCtx.getTransactionSubsystem().getLogManager());
} catch (Throwable th) {
throw new HyracksDataException(th);
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/953eb966/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/transactions/Resource.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/transactions/Resource.java b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/transactions/Resource.java
index e53ca7f..81cb58a 100644
--- a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/transactions/Resource.java
+++ b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/transactions/Resource.java
@@ -21,7 +21,7 @@ package org.apache.asterix.common.transactions;
import java.io.Serializable;
import java.util.List;
-import org.apache.hyracks.api.application.INCApplicationContext;
+import org.apache.hyracks.api.application.INCServiceContext;
import org.apache.hyracks.api.dataflow.value.IBinaryComparatorFactory;
import org.apache.hyracks.api.dataflow.value.ITypeTraits;
import org.apache.hyracks.api.exceptions.HyracksDataException;
@@ -72,7 +72,7 @@ public abstract class Resource implements Serializable {
return datasetId;
}
- public abstract ILSMIndex createIndexInstance(INCApplicationContext appCtx, LocalResource resource)
+ public abstract ILSMIndex createIndexInstance(INCServiceContext ncServiceCtx, LocalResource resource)
throws HyracksDataException;
public static int getIoDeviceNum(IIOManager ioManager, IODeviceHandle deviceHandle) {
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/953eb966/asterixdb/asterix-experiments/src/main/java/org/apache/asterix/experiment/builder/AbstractLocalExperimentBuilder.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-experiments/src/main/java/org/apache/asterix/experiment/builder/AbstractLocalExperimentBuilder.java b/asterixdb/asterix-experiments/src/main/java/org/apache/asterix/experiment/builder/AbstractLocalExperimentBuilder.java
index eac4ac2..315ad9c 100644
--- a/asterixdb/asterix-experiments/src/main/java/org/apache/asterix/experiment/builder/AbstractLocalExperimentBuilder.java
+++ b/asterixdb/asterix-experiments/src/main/java/org/apache/asterix/experiment/builder/AbstractLocalExperimentBuilder.java
@@ -19,19 +19,7 @@
package org.apache.asterix.experiment.builder;
-import java.nio.file.Path;
-import java.nio.file.Paths;
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.List;
-
import org.apache.asterix.experiment.action.base.SequentialActionList;
-import org.apache.asterix.hyracks.bootstrap.CCApplicationEntryPoint;
-import org.apache.asterix.hyracks.bootstrap.NCApplicationEntryPoint;
-import org.apache.hyracks.control.cc.ClusterControllerService;
-import org.apache.hyracks.control.common.controllers.CCConfig;
-import org.apache.hyracks.control.common.controllers.NCConfig;
-import org.apache.hyracks.control.nc.NodeControllerService;
public abstract class AbstractLocalExperimentBuilder extends AbstractExperimentBuilder {
@@ -46,74 +34,4 @@ public abstract class AbstractLocalExperimentBuilder extends AbstractExperimentB
protected abstract void addPost(SequentialActionList post);
-// @Override
-// protected void prePost(SequentialExecutableSet pre, SequentialExecutableSet post) {
-// int ccClientPort = 1098;
-// int ccClusterPort = 1099;
-// CCConfig ccConfig = new CCConfig();
-// ccConfig.clusterNetIpAddress = "127.0.0.1";
-// ccConfig.clientNetIpAddress = "127.0.0.1";
-// ccConfig.clientNetPort = ccClientPort;
-// ccConfig.clusterNetPort = ccClusterPort;
-// ccConfig.defaultMaxJobAttempts = 0;
-// ccConfig.resultTTL = 30000;
-// ccConfig.resultSweepThreshold = 1000;
-// ccConfig.appCCMainClass = CCApplicationEntryPoint.class.getName();
-// final ClusterControllerService cc;
-// try {
-// cc = new ClusterControllerService(ccConfig);
-// } catch (Exception e) {
-// throw new IllegalArgumentException(e);
-// }
-//
-// final List<NodeControllerService> ncs = new ArrayList<>();
-// for (int i = 0; i < nNodeControllers; ++i) {
-// NCConfig ncConfig = new NCConfig();
-// ncConfig.ccHost = "localhost";
-// ncConfig.ccPort = ccClusterPort;
-// ncConfig.clusterNetIPAddress = "127.0.0.1";
-// ncConfig.dataIPAddress = "127.0.0.1";
-// ncConfig.datasetIPAddress = "127.0.0.1";
-// ncConfig.nodeId = "nc" + String.valueOf((i + 1));
-// ncConfig.resultTTL = 30000;
-// ncConfig.resultSweepThreshold = 1000;
-// Path p0 = Paths.get(System.getProperty("java.io.tmpdir"), ncConfig.nodeId, "iodevice0");
-// Path p1 = Paths.get(System.getProperty("java.io.tmpdir"), ncConfig.nodeId, "iodevice1");
-// ncConfig.ioDevices = p0.toString() + "," + p1.toString();
-// ncConfig.appNCMainClass = NCApplicationEntryPoint.class.getName();
-// NodeControllerService nc;
-// try {
-// nc = new NodeControllerService(ncConfig);
-// } catch (Exception e) {
-// throw new IllegalArgumentException(e);
-// }
-// ncs.add(nc);
-// }
-//
-// pre.add(new AbstractExecutable() {
-//
-// @Override
-// protected void doExecute() throws Exception {
-// cc.start();
-// for (NodeControllerService nc : ncs) {
-// nc.start();
-// }
-// }
-// });
-//
-// post.add(new AbstractExecutable() {
-//
-// @Override
-// protected void doExecute() throws Exception {
-// Collections.reverse(ncs);
-// for (NodeControllerService nc : ncs) {
-// nc.stop();
-// }
-// cc.stop();
-// System.exit(1);
-// }
-// });
-// addPre(pre);
-// addPost(post);
-// }
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/953eb966/asterixdb/asterix-external-data/pom.xml
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-external-data/pom.xml b/asterixdb/asterix-external-data/pom.xml
index d1f67f0..4bb06c2 100644
--- a/asterixdb/asterix-external-data/pom.xml
+++ b/asterixdb/asterix-external-data/pom.xml
@@ -268,7 +268,6 @@
<dependency>
<groupId>org.apache.hive</groupId>
<artifactId>hive-serde</artifactId>
- <version>0.13.0</version>
</dependency>
<dependency>
<groupId>com.e-movimento.tinytools</groupId>
@@ -368,10 +367,6 @@
<artifactId>commons-lang3</artifactId>
</dependency>
<dependency>
- <groupId>org.apache.hyracks</groupId>
- <artifactId>hyracks-control-nc</artifactId>
- </dependency>
- <dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
</dependency>
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/953eb966/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/adapter/factory/GenericAdapterFactory.java
----------------------------------------------------------------------
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 37262b7..577da5e 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
@@ -85,10 +85,10 @@ public class GenericAdapterFactory implements IIndexingAdapterFactory, IAdapterF
@Override
public synchronized IDataSourceAdapter createAdapter(IHyracksTaskContext ctx, int partition)
throws HyracksDataException {
- IAppRuntimeContext runtimeCtx =
- (IAppRuntimeContext) ctx.getJobletContext().getApplicationContext().getApplicationObject();
+ IAppRuntimeContext appCtx =
+ (IAppRuntimeContext) ctx.getJobletContext().getServiceContext().getApplicationContext();
try {
- restoreExternalObjects(runtimeCtx.getLibraryManager());
+ restoreExternalObjects(appCtx.getLibraryManager());
} catch (Exception e) {
throw HyracksDataException.create(e);
}