asterixdb-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From mhub...@apache.org
Subject asterixdb git commit: [ASTERIXDB-2228][CLUS] IFaultToleranceStrategy -> INcLifecycleCoordinator
Date Sun, 07 Jan 2018 19:53:56 GMT
Repository: asterixdb
Updated Branches:
  refs/heads/master c47773778 -> f12288854


[ASTERIXDB-2228][CLUS] IFaultToleranceStrategy -> INcLifecycleCoordinator

- user model changes: no
- storage format changes: no
- interface changes: yes
  - Rename IFaultToleranceStrategy -> INcLifecycleCoordinator

Details:
- Rename IFaultToleranceStrategy since it doesn't provide
  fault tolerance anymore but rather coordinates node
  lifecycle.
- Make INcLifecycleCoordinator extensible.
- Remove unneeded FaultToleranceStrategyFactory.

Change-Id: I51a2d6f5402630683f709806e6c01f0c7e83914f
Reviewed-on: https://asterix-gerrit.ics.uci.edu/2266
Sonar-Qube: Jenkins <jenkins@fulliautomatix.ics.uci.edu>
Tested-by: Jenkins <jenkins@fulliautomatix.ics.uci.edu>
Integration-Tests: Jenkins <jenkins@fulliautomatix.ics.uci.edu>
Contrib: Jenkins <jenkins@fulliautomatix.ics.uci.edu>
Reviewed-by: Till Westmann <tillw@apache.org>


Project: http://git-wip-us.apache.org/repos/asf/asterixdb/repo
Commit: http://git-wip-us.apache.org/repos/asf/asterixdb/commit/f1228885
Tree: http://git-wip-us.apache.org/repos/asf/asterixdb/tree/f1228885
Diff: http://git-wip-us.apache.org/repos/asf/asterixdb/diff/f1228885

Branch: refs/heads/master
Commit: f1228885446eed8a49db1a674538376f9bb3f62c
Parents: c477737
Author: Murtadha Hubail <mhubail@apache.org>
Authored: Sun Jan 7 12:31:01 2018 +0300
Committer: Murtadha Hubail <mhubail@apache.org>
Committed: Sun Jan 7 11:53:29 2018 -0800

----------------------------------------------------------------------
 .../api/http/server/ClusterApiServlet.java      |   2 +-
 .../FaultToleranceStrategyFactory.java          |  38 ----
 .../app/replication/NcLifecycleCoordinator.java | 218 ++++++++++++++++++
 .../replication/NoFaultToleranceStrategy.java   | 221 -------------------
 .../message/MetadataNodeResponseMessage.java    |   2 +-
 .../message/NCLifecycleTaskReportMessage.java   |   2 +-
 .../RegistrationTasksRequestMessage.java        |   2 +-
 .../hyracks/bootstrap/CCApplication.java        |  19 +-
 .../runtime/ClusterStateManagerTest.java        |  10 +-
 .../common/dataflow/ICcApplicationContext.java  |   6 +-
 .../replication/IFaultToleranceStrategy.java    |  75 -------
 .../replication/INcLifecycleCoordinator.java    |  63 ++++++
 .../runtime/utils/CcApplicationContext.java     |   8 +-
 .../runtime/utils/ClusterStateManager.java      |  18 +-
 14 files changed, 313 insertions(+), 371 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/asterixdb/blob/f1228885/asterixdb/asterix-app/src/main/java/org/apache/asterix/api/http/server/ClusterApiServlet.java
----------------------------------------------------------------------
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 38ee10c..d239038 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
@@ -169,7 +169,7 @@ public class ClusterApiServlet extends AbstractServlet {
 
     private void processMetadataNode(IServletRequest request, IServletResponse response) throws HyracksDataException {
         final String node = request.getParameter("node");
-        appCtx.getFaultToleranceStrategy().notifyMetadataNodeChange(node);
+        appCtx.getNcLifecycleCoordinator().notifyMetadataNodeChange(node);
         response.setStatus(HttpResponseStatus.OK);
     }
 }

http://git-wip-us.apache.org/repos/asf/asterixdb/blob/f1228885/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/replication/FaultToleranceStrategyFactory.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/replication/FaultToleranceStrategyFactory.java b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/replication/FaultToleranceStrategyFactory.java
deleted file mode 100644
index 5a715d6..0000000
--- a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/replication/FaultToleranceStrategyFactory.java
+++ /dev/null
@@ -1,38 +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.app.replication;
-
-import org.apache.asterix.common.replication.IFaultToleranceStrategy;
-import org.apache.hyracks.api.application.ICCServiceContext;
-
-public class FaultToleranceStrategyFactory {
-
-    private FaultToleranceStrategyFactory() {
-        throw new AssertionError();
-    }
-
-    public static IFaultToleranceStrategy create(ICCServiceContext serviceCtx, boolean replicationEnabled) {
-        Class<? extends IFaultToleranceStrategy> clazz = NoFaultToleranceStrategy.class;
-        try {
-            return clazz.newInstance().from(serviceCtx, replicationEnabled);
-        } catch (InstantiationException | IllegalAccessException e) {
-            throw new IllegalStateException(e);
-        }
-    }
-}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/asterixdb/blob/f1228885/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/replication/NcLifecycleCoordinator.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/replication/NcLifecycleCoordinator.java b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/replication/NcLifecycleCoordinator.java
new file mode 100644
index 0000000..54b1a01
--- /dev/null
+++ b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/replication/NcLifecycleCoordinator.java
@@ -0,0 +1,218 @@
+/*
+ * 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.app.replication;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+import org.apache.asterix.app.nc.task.BindMetadataNodeTask;
+import org.apache.asterix.app.nc.task.CheckpointTask;
+import org.apache.asterix.app.nc.task.ExternalLibrarySetupTask;
+import org.apache.asterix.app.nc.task.LocalRecoveryTask;
+import org.apache.asterix.app.nc.task.MetadataBootstrapTask;
+import org.apache.asterix.app.nc.task.ReportLocalCountersTask;
+import org.apache.asterix.app.nc.task.StartLifecycleComponentsTask;
+import org.apache.asterix.app.nc.task.StartReplicationServiceTask;
+import org.apache.asterix.app.replication.message.MetadataNodeRequestMessage;
+import org.apache.asterix.app.replication.message.MetadataNodeResponseMessage;
+import org.apache.asterix.app.replication.message.NCLifecycleTaskReportMessage;
+import org.apache.asterix.app.replication.message.RegistrationTasksRequestMessage;
+import org.apache.asterix.app.replication.message.RegistrationTasksResponseMessage;
+import org.apache.asterix.common.api.INCLifecycleTask;
+import org.apache.asterix.common.cluster.ClusterPartition;
+import org.apache.asterix.common.cluster.IClusterStateManager;
+import org.apache.asterix.common.exceptions.ErrorCode;
+import org.apache.asterix.common.exceptions.RuntimeDataException;
+import org.apache.asterix.common.messaging.api.ICCMessageBroker;
+import org.apache.asterix.common.replication.INCLifecycleMessage;
+import org.apache.asterix.common.replication.INcLifecycleCoordinator;
+import org.apache.asterix.common.transactions.IRecoveryManager.SystemState;
+import org.apache.asterix.metadata.MetadataManager;
+import org.apache.hyracks.api.application.ICCServiceContext;
+import org.apache.hyracks.api.client.NodeStatus;
+import org.apache.hyracks.api.exceptions.HyracksDataException;
+import org.apache.logging.log4j.Level;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+
+public class NcLifecycleCoordinator implements INcLifecycleCoordinator {
+
+    private static final Logger LOGGER = LogManager.getLogger();
+    protected IClusterStateManager clusterManager;
+    protected String metadataNodeId;
+    protected Set<String> pendingStartupCompletionNodes = new HashSet<>();
+    protected final ICCMessageBroker messageBroker;
+    private final boolean replicationEnabled;
+
+    public NcLifecycleCoordinator(ICCServiceContext serviceCtx, boolean replicationEnabled) {
+        this.messageBroker = (ICCMessageBroker) serviceCtx.getMessageBroker();
+        this.replicationEnabled = replicationEnabled;
+    }
+
+    @Override
+    public void notifyNodeJoin(String nodeId) {
+        pendingStartupCompletionNodes.add(nodeId);
+    }
+
+    @Override
+    public void notifyNodeFailure(String nodeId) throws HyracksDataException {
+        pendingStartupCompletionNodes.remove(nodeId);
+        clusterManager.updateNodePartitions(nodeId, false);
+        if (nodeId.equals(metadataNodeId)) {
+            clusterManager.updateMetadataNode(metadataNodeId, false);
+        }
+        clusterManager.refreshState();
+    }
+
+    @Override
+    public void process(INCLifecycleMessage message) throws HyracksDataException {
+        switch (message.getType()) {
+            case REGISTRATION_TASKS_REQUEST:
+                process((RegistrationTasksRequestMessage) message);
+                break;
+            case REGISTRATION_TASKS_RESULT:
+                process((NCLifecycleTaskReportMessage) message);
+                break;
+            case METADATA_NODE_RESPONSE:
+                process((MetadataNodeResponseMessage) message);
+                break;
+            default:
+                throw new RuntimeDataException(ErrorCode.UNSUPPORTED_MESSAGE_TYPE, message.getType().name());
+        }
+    }
+
+    @Override
+    public void bindTo(IClusterStateManager clusterManager) {
+        this.clusterManager = clusterManager;
+        metadataNodeId = clusterManager.getCurrentMetadataNodeId();
+    }
+
+    private void process(RegistrationTasksRequestMessage msg) throws HyracksDataException {
+        final String nodeId = msg.getNodeId();
+        List<INCLifecycleTask> tasks = buildNCRegTasks(msg.getNodeId(), msg.getNodeStatus(), msg.getState());
+        RegistrationTasksResponseMessage response = new RegistrationTasksResponseMessage(nodeId, tasks);
+        try {
+            messageBroker.sendApplicationMessageToNC(response, msg.getNodeId());
+        } catch (Exception e) {
+            throw HyracksDataException.create(e);
+        }
+    }
+
+    private void process(NCLifecycleTaskReportMessage msg) throws HyracksDataException {
+        pendingStartupCompletionNodes.remove(msg.getNodeId());
+        if (msg.isSuccess()) {
+            clusterManager.updateNodePartitions(msg.getNodeId(), true);
+            if (msg.getNodeId().equals(metadataNodeId)) {
+                clusterManager.updateMetadataNode(metadataNodeId, true);
+            }
+            clusterManager.refreshState();
+        } else {
+            if (LOGGER.isErrorEnabled()) {
+                LOGGER.log(Level.ERROR, msg.getNodeId() + " failed to complete startup. ", msg.getException());
+            }
+        }
+    }
+
+    protected List<INCLifecycleTask> buildNCRegTasks(String nodeId, NodeStatus nodeStatus, SystemState state) {
+        LOGGER.log(Level.INFO, () -> "Building registration tasks for node: " + nodeId + " with state: " + state);
+        final boolean isMetadataNode = nodeId.equals(metadataNodeId);
+        if (nodeStatus == NodeStatus.ACTIVE) {
+            /*
+             * if the node state is already ACTIVE then it completed
+             * booting and just re-registering with a new/failed CC.
+             */
+            return buildActiveNCRegTasks(isMetadataNode);
+        }
+        final List<INCLifecycleTask> tasks = new ArrayList<>();
+        if (state == SystemState.CORRUPTED) {
+            //need to perform local recovery for node partitions
+            LocalRecoveryTask rt = new LocalRecoveryTask(
+                    Arrays.asList(clusterManager.getNodePartitions(nodeId)).stream()
+                            .map(ClusterPartition::getPartitionId).collect(Collectors.toSet()));
+            tasks.add(rt);
+        }
+        if (replicationEnabled) {
+            tasks.add(new StartReplicationServiceTask());
+        }
+        if (isMetadataNode) {
+            tasks.add(new MetadataBootstrapTask());
+        }
+        tasks.add(new ExternalLibrarySetupTask(isMetadataNode));
+        tasks.add(new ReportLocalCountersTask());
+        tasks.add(new CheckpointTask());
+        tasks.add(new StartLifecycleComponentsTask());
+        if (isMetadataNode) {
+            tasks.add(new BindMetadataNodeTask(true));
+        }
+        return tasks;
+    }
+
+    protected List<INCLifecycleTask> buildActiveNCRegTasks(boolean metadataNode) {
+        final List<INCLifecycleTask> tasks = new ArrayList<>();
+        if (metadataNode) {
+            // need to unbind from old distributed state then rebind to new one
+            tasks.add(new BindMetadataNodeTask(false));
+            tasks.add(new BindMetadataNodeTask(true));
+        }
+        tasks.add(new ReportLocalCountersTask());
+        return tasks;
+    }
+
+    @Override
+    public void notifyMetadataNodeChange(String node) throws HyracksDataException {
+        if (metadataNodeId.equals(node)) {
+            return;
+        }
+        // if current metadata node is active, we need to unbind its metadata proxy object
+        if (clusterManager.isMetadataNodeActive()) {
+            MetadataNodeRequestMessage msg = new MetadataNodeRequestMessage(false);
+            try {
+                messageBroker.sendApplicationMessageToNC(msg, metadataNodeId);
+                // when the current node responses, we will bind to the new one
+                metadataNodeId = node;
+            } catch (Exception e) {
+                throw HyracksDataException.create(e);
+            }
+        } else {
+            requestMetadataNodeTakeover(node);
+        }
+    }
+
+    private void process(MetadataNodeResponseMessage response) throws HyracksDataException {
+        // rebind metadata node since it might be changing
+        MetadataManager.INSTANCE.rebindMetadataNode();
+        clusterManager.updateMetadataNode(response.getNodeId(), response.isExported());
+        if (!response.isExported()) {
+            requestMetadataNodeTakeover(metadataNodeId);
+        }
+    }
+
+    private void requestMetadataNodeTakeover(String node) throws HyracksDataException {
+        MetadataNodeRequestMessage msg = new MetadataNodeRequestMessage(true);
+        try {
+            messageBroker.sendApplicationMessageToNC(msg, node);
+        } catch (Exception e) {
+            throw HyracksDataException.create(e);
+        }
+    }
+}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/asterixdb/blob/f1228885/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/replication/NoFaultToleranceStrategy.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/replication/NoFaultToleranceStrategy.java b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/replication/NoFaultToleranceStrategy.java
deleted file mode 100644
index 49f4b32..0000000
--- a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/replication/NoFaultToleranceStrategy.java
+++ /dev/null
@@ -1,221 +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.app.replication;
-
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.HashSet;
-import java.util.List;
-import java.util.Set;
-import java.util.stream.Collectors;
-
-import org.apache.asterix.app.nc.task.BindMetadataNodeTask;
-import org.apache.asterix.app.nc.task.CheckpointTask;
-import org.apache.asterix.app.nc.task.ExternalLibrarySetupTask;
-import org.apache.asterix.app.nc.task.LocalRecoveryTask;
-import org.apache.asterix.app.nc.task.MetadataBootstrapTask;
-import org.apache.asterix.app.nc.task.ReportLocalCountersTask;
-import org.apache.asterix.app.nc.task.StartLifecycleComponentsTask;
-import org.apache.asterix.app.nc.task.StartReplicationServiceTask;
-import org.apache.asterix.app.replication.message.MetadataNodeRequestMessage;
-import org.apache.asterix.app.replication.message.MetadataNodeResponseMessage;
-import org.apache.asterix.app.replication.message.NCLifecycleTaskReportMessage;
-import org.apache.asterix.app.replication.message.RegistrationTasksRequestMessage;
-import org.apache.asterix.app.replication.message.RegistrationTasksResponseMessage;
-import org.apache.asterix.common.api.INCLifecycleTask;
-import org.apache.asterix.common.cluster.ClusterPartition;
-import org.apache.asterix.common.cluster.IClusterStateManager;
-import org.apache.asterix.common.exceptions.ErrorCode;
-import org.apache.asterix.common.exceptions.RuntimeDataException;
-import org.apache.asterix.common.messaging.api.ICCMessageBroker;
-import org.apache.asterix.common.replication.IFaultToleranceStrategy;
-import org.apache.asterix.common.replication.INCLifecycleMessage;
-import org.apache.asterix.common.transactions.IRecoveryManager.SystemState;
-import org.apache.asterix.metadata.MetadataManager;
-import org.apache.hyracks.api.application.ICCServiceContext;
-import org.apache.hyracks.api.client.NodeStatus;
-import org.apache.hyracks.api.exceptions.HyracksDataException;
-import org.apache.logging.log4j.Level;
-import org.apache.logging.log4j.LogManager;
-import org.apache.logging.log4j.Logger;
-
-public class NoFaultToleranceStrategy implements IFaultToleranceStrategy {
-
-    private static final Logger LOGGER = LogManager.getLogger();
-    private IClusterStateManager clusterManager;
-    private String metadataNodeId;
-    private Set<String> pendingStartupCompletionNodes = new HashSet<>();
-    private ICCMessageBroker messageBroker;
-    private boolean replicationEnabled;
-
-    @Override
-    public void notifyNodeJoin(String nodeId) throws HyracksDataException {
-        pendingStartupCompletionNodes.add(nodeId);
-    }
-
-    @Override
-    public void notifyNodeFailure(String nodeId) throws HyracksDataException {
-        pendingStartupCompletionNodes.remove(nodeId);
-        clusterManager.updateNodePartitions(nodeId, false);
-        if (nodeId.equals(metadataNodeId)) {
-            clusterManager.updateMetadataNode(metadataNodeId, false);
-        }
-        clusterManager.refreshState();
-    }
-
-    @Override
-    public void process(INCLifecycleMessage message) throws HyracksDataException {
-        switch (message.getType()) {
-            case REGISTRATION_TASKS_REQUEST:
-                process((RegistrationTasksRequestMessage) message);
-                break;
-            case REGISTRATION_TASKS_RESULT:
-                process((NCLifecycleTaskReportMessage) message);
-                break;
-            case METADATA_NODE_RESPONSE:
-                process((MetadataNodeResponseMessage) message);
-                break;
-            default:
-                throw new RuntimeDataException(ErrorCode.UNSUPPORTED_MESSAGE_TYPE, message.getType().name());
-        }
-    }
-
-    @Override
-    public IFaultToleranceStrategy from(ICCServiceContext serviceCtx, boolean replicationEnabled) {
-        NoFaultToleranceStrategy ft = new NoFaultToleranceStrategy();
-        ft.messageBroker = (ICCMessageBroker) serviceCtx.getMessageBroker();
-        ft.replicationEnabled = replicationEnabled;
-        return ft;
-    }
-
-    @Override
-    public void bindTo(IClusterStateManager clusterManager) {
-        this.clusterManager = clusterManager;
-        metadataNodeId = clusterManager.getCurrentMetadataNodeId();
-    }
-
-    private void process(RegistrationTasksRequestMessage msg) throws HyracksDataException {
-        final String nodeId = msg.getNodeId();
-        List<INCLifecycleTask> tasks = buildNCRegTasks(msg.getNodeId(), msg.getNodeStatus(), msg.getState());
-        RegistrationTasksResponseMessage response = new RegistrationTasksResponseMessage(nodeId, tasks);
-        try {
-            messageBroker.sendApplicationMessageToNC(response, msg.getNodeId());
-        } catch (Exception e) {
-            throw HyracksDataException.create(e);
-        }
-    }
-
-    private void process(NCLifecycleTaskReportMessage msg) throws HyracksDataException {
-        pendingStartupCompletionNodes.remove(msg.getNodeId());
-        if (msg.isSuccess()) {
-            clusterManager.updateNodePartitions(msg.getNodeId(), true);
-            if (msg.getNodeId().equals(metadataNodeId)) {
-                clusterManager.updateMetadataNode(metadataNodeId, true);
-            }
-            clusterManager.refreshState();
-        } else {
-            if (LOGGER.isErrorEnabled()) {
-                LOGGER.log(Level.ERROR, msg.getNodeId() + " failed to complete startup. ", msg.getException());
-            }
-        }
-    }
-
-    private List<INCLifecycleTask> buildNCRegTasks(String nodeId, NodeStatus nodeStatus, SystemState state) {
-        LOGGER.log(Level.INFO, () -> "Building registration tasks for node: " + nodeId + " with state: " + state);
-        final boolean isMetadataNode = nodeId.equals(metadataNodeId);
-        if (nodeStatus == NodeStatus.ACTIVE) {
-            /*
-             * if the node state is already ACTIVE then it completed
-             * booting and just re-registering with a new/failed CC.
-             */
-            return buildActiveNCRegTasks(isMetadataNode);
-        }
-        final List<INCLifecycleTask> tasks = new ArrayList<>();
-        if (state == SystemState.CORRUPTED) {
-            //need to perform local recovery for node partitions
-            LocalRecoveryTask rt = new LocalRecoveryTask(
-                    Arrays.asList(clusterManager.getNodePartitions(nodeId)).stream()
-                            .map(ClusterPartition::getPartitionId).collect(Collectors.toSet()));
-            tasks.add(rt);
-        }
-        if (replicationEnabled) {
-            tasks.add(new StartReplicationServiceTask());
-        }
-        if (isMetadataNode) {
-            tasks.add(new MetadataBootstrapTask());
-        }
-        tasks.add(new ExternalLibrarySetupTask(isMetadataNode));
-        tasks.add(new ReportLocalCountersTask());
-        tasks.add(new CheckpointTask());
-        tasks.add(new StartLifecycleComponentsTask());
-        if (isMetadataNode) {
-            tasks.add(new BindMetadataNodeTask(true));
-        }
-        return tasks;
-    }
-
-    private List<INCLifecycleTask> buildActiveNCRegTasks(boolean metadataNode) {
-        final List<INCLifecycleTask> tasks = new ArrayList<>();
-        if (metadataNode) {
-            // need to unbind from old distributed state then rebind to new one
-            tasks.add(new BindMetadataNodeTask(false));
-            tasks.add(new BindMetadataNodeTask(true));
-        }
-        tasks.add(new ReportLocalCountersTask());
-        return tasks;
-    }
-
-    @Override
-    public void notifyMetadataNodeChange(String node) throws HyracksDataException {
-        if (metadataNodeId.equals(node)) {
-            return;
-        }
-        // if current metadata node is active, we need to unbind its metadata proxy object
-        if (clusterManager.isMetadataNodeActive()) {
-            MetadataNodeRequestMessage msg = new MetadataNodeRequestMessage(false);
-            try {
-                messageBroker.sendApplicationMessageToNC(msg, metadataNodeId);
-                // when the current node responses, we will bind to the new one
-                metadataNodeId = node;
-            } catch (Exception e) {
-                throw HyracksDataException.create(e);
-            }
-        } else {
-            requestMetadataNodeTakeover(node);
-        }
-    }
-
-    private void process(MetadataNodeResponseMessage response) throws HyracksDataException {
-        // rebind metadata node since it might be changing
-        MetadataManager.INSTANCE.rebindMetadataNode();
-        clusterManager.updateMetadataNode(response.getNodeId(), response.isExported());
-        if (!response.isExported()) {
-            requestMetadataNodeTakeover(metadataNodeId);
-        }
-    }
-
-    private void requestMetadataNodeTakeover(String node) throws HyracksDataException {
-        MetadataNodeRequestMessage msg = new MetadataNodeRequestMessage(true);
-        try {
-            messageBroker.sendApplicationMessageToNC(msg, node);
-        } catch (Exception e) {
-            throw HyracksDataException.create(e);
-        }
-    }
-}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/asterixdb/blob/f1228885/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/replication/message/MetadataNodeResponseMessage.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/replication/message/MetadataNodeResponseMessage.java b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/replication/message/MetadataNodeResponseMessage.java
index ebde9b9..815d878 100644
--- a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/replication/message/MetadataNodeResponseMessage.java
+++ b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/replication/message/MetadataNodeResponseMessage.java
@@ -41,7 +41,7 @@ public class MetadataNodeResponseMessage implements INCLifecycleMessage, ICcAddr
 
     @Override
     public void handle(ICcApplicationContext appCtx) throws HyracksDataException, InterruptedException {
-        ((CcApplicationContext) appCtx).getFaultToleranceStrategy().process(this);
+        ((CcApplicationContext) appCtx).getNcLifecycleCoordinator().process(this);
     }
 
     @Override

http://git-wip-us.apache.org/repos/asf/asterixdb/blob/f1228885/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/replication/message/NCLifecycleTaskReportMessage.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/replication/message/NCLifecycleTaskReportMessage.java b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/replication/message/NCLifecycleTaskReportMessage.java
index b654fd8..5ac3a0c 100644
--- a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/replication/message/NCLifecycleTaskReportMessage.java
+++ b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/replication/message/NCLifecycleTaskReportMessage.java
@@ -37,7 +37,7 @@ public class NCLifecycleTaskReportMessage implements INCLifecycleMessage, ICcAdd
 
     @Override
     public void handle(ICcApplicationContext appCtx) throws HyracksDataException, InterruptedException {
-        appCtx.getFaultToleranceStrategy().process(this);
+        appCtx.getNcLifecycleCoordinator().process(this);
     }
 
     public String getNodeId() {

http://git-wip-us.apache.org/repos/asf/asterixdb/blob/f1228885/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/replication/message/RegistrationTasksRequestMessage.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/replication/message/RegistrationTasksRequestMessage.java b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/replication/message/RegistrationTasksRequestMessage.java
index b60ef8b..6ca576a 100644
--- a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/replication/message/RegistrationTasksRequestMessage.java
+++ b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/replication/message/RegistrationTasksRequestMessage.java
@@ -58,7 +58,7 @@ public class RegistrationTasksRequestMessage implements INCLifecycleMessage, ICc
 
     @Override
     public void handle(ICcApplicationContext appCtx) throws HyracksDataException, InterruptedException {
-        appCtx.getFaultToleranceStrategy().process(this);
+        appCtx.getNcLifecycleCoordinator().process(this);
     }
 
     public SystemState getState() {

http://git-wip-us.apache.org/repos/asf/asterixdb/blob/f1228885/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
index c2a0fc1..11f4e1c 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
@@ -55,7 +55,7 @@ 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.external.ExternalLibraryUtils;
-import org.apache.asterix.app.replication.FaultToleranceStrategyFactory;
+import org.apache.asterix.app.replication.NcLifecycleCoordinator;
 import org.apache.asterix.common.api.AsterixThreadFactory;
 import org.apache.asterix.common.api.IClusterManagementWork;
 import org.apache.asterix.common.api.INodeJobTracker;
@@ -67,9 +67,7 @@ import org.apache.asterix.common.config.ReplicationProperties;
 import org.apache.asterix.common.context.IStorageComponentProvider;
 import org.apache.asterix.common.dataflow.ICcApplicationContext;
 import org.apache.asterix.common.library.ILibraryManager;
-import org.apache.asterix.common.replication.IFaultToleranceStrategy;
-import org.apache.asterix.common.replication.IReplicationStrategy;
-import org.apache.asterix.common.replication.ReplicationStrategyFactory;
+import org.apache.asterix.common.replication.INcLifecycleCoordinator;
 import org.apache.asterix.common.utils.Servlets;
 import org.apache.asterix.external.library.ExternalLibraryManager;
 import org.apache.asterix.file.StorageComponentProvider;
@@ -142,13 +140,12 @@ public class CCApplication extends BaseCCApplication {
         ILibraryManager libraryManager = new ExternalLibraryManager();
         ReplicationProperties repProp = new ReplicationProperties(
                 PropertiesAccessor.getInstance(ccServiceCtx.getAppConfig()));
-        IFaultToleranceStrategy ftStrategy =
-                FaultToleranceStrategyFactory.create(ccServiceCtx, repProp.isReplicationEnabled());
+        INcLifecycleCoordinator lifecycleCoordinator = createNcLifeCycleCoordinator(repProp.isReplicationEnabled());
         ExternalLibraryUtils.setUpExternaLibraries(libraryManager, false);
         componentProvider = new StorageComponentProvider();
         GlobalRecoveryManager globalRecoveryManager = createGlobalRecoveryManager();
         statementExecutorCtx = new StatementExecutorContext();
-        appCtx = createApplicationContext(libraryManager, globalRecoveryManager, ftStrategy);
+        appCtx = createApplicationContext(libraryManager, globalRecoveryManager, lifecycleCoordinator);
         List<AsterixExtension> extensions = new ArrayList<>();
         extensions.addAll(this.getExtensions());
         ccExtensionManager = new CCExtensionManager(extensions);
@@ -177,10 +174,10 @@ public class CCApplication extends BaseCCApplication {
     }
 
     protected ICcApplicationContext createApplicationContext(ILibraryManager libraryManager,
-            GlobalRecoveryManager globalRecoveryManager, IFaultToleranceStrategy ftStrategy)
+            GlobalRecoveryManager globalRecoveryManager, INcLifecycleCoordinator lifecycleCoordinator)
             throws AlgebricksException, IOException {
         return new CcApplicationContext(ccServiceCtx, getHcc(), libraryManager, () -> MetadataManager.INSTANCE,
-                globalRecoveryManager, ftStrategy, new ActiveNotificationHandler(), componentProvider,
+                globalRecoveryManager, lifecycleCoordinator, new ActiveNotificationHandler(), componentProvider,
                 new MetadataLockManager());
     }
 
@@ -188,6 +185,10 @@ public class CCApplication extends BaseCCApplication {
         return new GlobalRecoveryManager(ccServiceCtx, getHcc(), componentProvider);
     }
 
+    protected INcLifecycleCoordinator createNcLifeCycleCoordinator(boolean replicationEnabled) {
+        return new NcLifecycleCoordinator(ccServiceCtx, replicationEnabled);
+    }
+
     @Override
     protected void configureLoggingLevel(Level level) {
         super.configureLoggingLevel(level);

http://git-wip-us.apache.org/repos/asf/asterixdb/blob/f1228885/asterixdb/asterix-app/src/test/java/org/apache/asterix/runtime/ClusterStateManagerTest.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-app/src/test/java/org/apache/asterix/runtime/ClusterStateManagerTest.java b/asterixdb/asterix-app/src/test/java/org/apache/asterix/runtime/ClusterStateManagerTest.java
index 07d3584..ca314aa 100644
--- a/asterixdb/asterix-app/src/test/java/org/apache/asterix/runtime/ClusterStateManagerTest.java
+++ b/asterixdb/asterix-app/src/test/java/org/apache/asterix/runtime/ClusterStateManagerTest.java
@@ -24,7 +24,7 @@ import java.util.SortedMap;
 import java.util.TreeMap;
 import java.util.concurrent.ConcurrentHashMap;
 
-import org.apache.asterix.app.replication.NoFaultToleranceStrategy;
+import org.apache.asterix.app.replication.NcLifecycleCoordinator;
 import org.apache.asterix.app.replication.message.NCLifecycleTaskReportMessage;
 import org.apache.asterix.common.api.IClusterManagementWork.ClusterState;
 import org.apache.asterix.common.cluster.ClusterPartition;
@@ -196,7 +196,7 @@ public class ClusterStateManagerTest {
             throws HyracksDataException {
         NCLifecycleTaskReportMessage msg = new NCLifecycleTaskReportMessage(nodeId, true);
         applicationContext.getResourceIdManager().report(nodeId, 0);
-        applicationContext.getFaultToleranceStrategy().process(msg);
+        applicationContext.getNcLifecycleCoordinator().process(msg);
     }
 
     private CcApplicationContext ccAppContext(ClusterStateManager csm) throws HyracksDataException {
@@ -207,9 +207,9 @@ public class ClusterStateManagerTest {
         Mockito.when(iccServiceContext.getAppConfig()).thenReturn(applicationConfig);
         Mockito.when(ccApplicationContext.getServiceContext()).thenReturn(iccServiceContext);
 
-        NoFaultToleranceStrategy fts = new NoFaultToleranceStrategy();
-        fts.bindTo(csm);
-        Mockito.when(ccApplicationContext.getFaultToleranceStrategy()).thenReturn(fts);
+        NcLifecycleCoordinator coordinator = new NcLifecycleCoordinator(ccApplicationContext.getServiceContext(), false);
+        coordinator.bindTo(csm);
+        Mockito.when(ccApplicationContext.getNcLifecycleCoordinator()).thenReturn(coordinator);
 
         MetadataProperties metadataProperties = mockMetadataProperties();
         Mockito.when(ccApplicationContext.getMetadataProperties()).thenReturn(metadataProperties);

http://git-wip-us.apache.org/repos/asf/asterixdb/blob/f1228885/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/dataflow/ICcApplicationContext.java
----------------------------------------------------------------------
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 6181ade..e02482d 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
@@ -26,7 +26,7 @@ import org.apache.asterix.common.cluster.IGlobalRecoveryManager;
 import org.apache.asterix.common.config.ExtensionProperties;
 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.replication.INcLifecycleCoordinator;
 import org.apache.asterix.common.transactions.IResourceIdManager;
 import org.apache.hyracks.api.application.ICCServiceContext;
 import org.apache.hyracks.api.job.IJobLifecycleListener;
@@ -62,9 +62,9 @@ public interface ICcApplicationContext extends IApplicationContext {
     IGlobalRecoveryManager getGlobalRecoveryManager();
 
     /**
-     * @return the fault tolerance strategy in use for the cluster
+     * @return the NC lifecycle coordinator in use for the cluster
      */
-    IFaultToleranceStrategy getFaultToleranceStrategy();
+    INcLifecycleCoordinator getNcLifecycleCoordinator();
 
     /**
      * @return the active notification handler at Cluster controller

http://git-wip-us.apache.org/repos/asf/asterixdb/blob/f1228885/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/replication/IFaultToleranceStrategy.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/replication/IFaultToleranceStrategy.java b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/replication/IFaultToleranceStrategy.java
deleted file mode 100644
index bf4cac9..0000000
--- a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/replication/IFaultToleranceStrategy.java
+++ /dev/null
@@ -1,75 +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.common.replication;
-
-import org.apache.asterix.common.cluster.IClusterStateManager;
-import org.apache.hyracks.api.application.ICCServiceContext;
-import org.apache.hyracks.api.exceptions.HyracksDataException;
-
-public interface IFaultToleranceStrategy {
-
-    /**
-     * Defines the logic of a {@link IFaultToleranceStrategy} when a node joins the cluster.
-     *
-     * @param nodeId
-     * @throws HyracksDataException
-     */
-    void notifyNodeJoin(String nodeId) throws HyracksDataException;
-
-    /**
-     * Defines the logic of a {@link IFaultToleranceStrategy} when a node leaves the cluster.
-     *
-     * @param nodeId
-     * @throws HyracksDataException
-     */
-    void notifyNodeFailure(String nodeId) throws HyracksDataException;
-
-    /**
-     * Binds the fault tolerance strategy to {@code cluserManager}.
-     *
-     * @param clusterManager
-     */
-    void bindTo(IClusterStateManager clusterManager);
-
-    /**
-     * Processes {@code message} based on the message type.
-     *
-     * @param message
-     * @throws HyracksDataException
-     */
-    void process(INCLifecycleMessage message) throws HyracksDataException;
-
-    /**
-     * Constructs a fault tolerance strategy.
-     *
-     * @param serviceCtx
-     * @param replicationEnabled
-     * @return the fault tolerance strategy
-     */
-    IFaultToleranceStrategy from(ICCServiceContext serviceCtx, boolean replicationEnabled);
-
-    /**
-     * Performs the required steps to change the metadata node to {@code node}
-     *
-     * @param node
-     */
-    default void notifyMetadataNodeChange(String node) throws HyracksDataException {
-        throw new UnsupportedOperationException(getClass() + " does not support metadata node change");
-    }
-}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/asterixdb/blob/f1228885/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/replication/INcLifecycleCoordinator.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/replication/INcLifecycleCoordinator.java b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/replication/INcLifecycleCoordinator.java
new file mode 100644
index 0000000..1a7c3c8
--- /dev/null
+++ b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/replication/INcLifecycleCoordinator.java
@@ -0,0 +1,63 @@
+/*
+ * 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.common.replication;
+
+import org.apache.asterix.common.cluster.IClusterStateManager;
+import org.apache.hyracks.api.exceptions.HyracksDataException;
+
+public interface INcLifecycleCoordinator {
+
+    /**
+     * Defines the logic of a {@link INcLifecycleCoordinator} when a node joins the cluster.
+     *
+     * @param nodeId
+     * @throws HyracksDataException
+     */
+    void notifyNodeJoin(String nodeId) throws HyracksDataException;
+
+    /**
+     * Defines the logic of a {@link INcLifecycleCoordinator} when a node leaves the cluster.
+     *
+     * @param nodeId
+     * @throws HyracksDataException
+     */
+    void notifyNodeFailure(String nodeId) throws HyracksDataException;
+
+    /**
+     * Binds the coordinator to {@code cluserManager}.
+     *
+     * @param clusterManager
+     */
+    void bindTo(IClusterStateManager clusterManager);
+
+    /**
+     * Processes {@code message} based on the message type.
+     *
+     * @param message
+     * @throws HyracksDataException
+     */
+    void process(INCLifecycleMessage message) throws HyracksDataException;
+
+    /**
+     * Performs the required steps to change the metadata node to {@code node}
+     *
+     * @param node
+     */
+    void notifyMetadataNodeChange(String node) throws HyracksDataException;
+}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/asterixdb/blob/f1228885/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/utils/CcApplicationContext.java
----------------------------------------------------------------------
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 f991cd4..d8f14a2 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
@@ -42,7 +42,7 @@ import org.apache.asterix.common.context.IStorageComponentProvider;
 import org.apache.asterix.common.dataflow.ICcApplicationContext;
 import org.apache.asterix.common.library.ILibraryManager;
 import org.apache.asterix.common.metadata.IMetadataBootstrap;
-import org.apache.asterix.common.replication.IFaultToleranceStrategy;
+import org.apache.asterix.common.replication.INcLifecycleCoordinator;
 import org.apache.asterix.common.transactions.IResourceIdManager;
 import org.apache.asterix.runtime.job.listener.NodeJobTracker;
 import org.apache.asterix.runtime.transaction.ResourceIdManager;
@@ -80,7 +80,7 @@ public class CcApplicationContext implements ICcApplicationContext {
     private Supplier<IMetadataBootstrap> metadataBootstrapSupplier;
     private IHyracksClientConnection hcc;
     private Object extensionManager;
-    private IFaultToleranceStrategy ftStrategy;
+    private INcLifecycleCoordinator ftStrategy;
     private IJobLifecycleListener activeLifeCycleListener;
     private IMetadataLockManager mdLockManager;
     private IClusterStateManager clusterStateManager;
@@ -88,7 +88,7 @@ public class CcApplicationContext implements ICcApplicationContext {
 
     public CcApplicationContext(ICCServiceContext ccServiceCtx, IHyracksClientConnection hcc,
             ILibraryManager libraryManager, Supplier<IMetadataBootstrap> metadataBootstrapSupplier,
-            IGlobalRecoveryManager globalRecoveryManager, IFaultToleranceStrategy ftStrategy,
+            IGlobalRecoveryManager globalRecoveryManager, INcLifecycleCoordinator ftStrategy,
             IJobLifecycleListener activeLifeCycleListener, IStorageComponentProvider storageComponentProvider,
             IMetadataLockManager mdLockManager) throws AlgebricksException, IOException {
         this.ccServiceCtx = ccServiceCtx;
@@ -232,7 +232,7 @@ public class CcApplicationContext implements ICcApplicationContext {
     }
 
     @Override
-    public IFaultToleranceStrategy getFaultToleranceStrategy() {
+    public INcLifecycleCoordinator getNcLifecycleCoordinator() {
         return ftStrategy;
     }
 

http://git-wip-us.apache.org/repos/asf/asterixdb/blob/f1228885/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/utils/ClusterStateManager.java
----------------------------------------------------------------------
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 c07200a..6994862 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
@@ -23,7 +23,6 @@ import java.util.Arrays;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.HashSet;
-import java.util.List;
 import java.util.Map;
 import java.util.Set;
 import java.util.SortedMap;
@@ -35,7 +34,7 @@ import org.apache.asterix.common.cluster.ClusterPartition;
 import org.apache.asterix.common.cluster.IClusterStateManager;
 import org.apache.asterix.common.exceptions.AsterixException;
 import org.apache.asterix.common.exceptions.ErrorCode;
-import org.apache.asterix.common.replication.IFaultToleranceStrategy;
+import org.apache.asterix.common.replication.INcLifecycleCoordinator;
 import org.apache.asterix.common.dataflow.ICcApplicationContext;
 import org.apache.asterix.common.transactions.IResourceIdManager;
 import org.apache.hyracks.algebricks.common.constraints.AlgebricksAbsolutePartitionConstraint;
@@ -60,11 +59,6 @@ import com.fasterxml.jackson.databind.node.ObjectNode;
  */
 
 public class ClusterStateManager implements IClusterStateManager {
-    /*
-     * TODO: currently after instance restarts we require all nodes to join again,
-     * otherwise the cluster wont be ACTIVE. we may overcome this by storing the cluster state before the instance
-     * shutdown and using it on startup to identify the nodes that are expected the join.
-     */
 
     private static final Logger LOGGER = LogManager.getLogger();
     private final Map<String, Map<IOption, Object>> ncConfigMap = new HashMap<>();
@@ -77,7 +71,7 @@ public class ClusterStateManager implements IClusterStateManager {
     private boolean metadataNodeActive = false;
     private Set<String> failedNodes = new HashSet<>();
     private Set<String> participantNodes = new HashSet<>();
-    private IFaultToleranceStrategy ftStrategy;
+    private INcLifecycleCoordinator lifecycleCoordinator;
     private ICcApplicationContext appCtx;
 
     @Override
@@ -86,8 +80,8 @@ public class ClusterStateManager implements IClusterStateManager {
         node2PartitionsMap = appCtx.getMetadataProperties().getNodePartitions();
         clusterPartitions = appCtx.getMetadataProperties().getClusterPartitions();
         currentMetadataNode = appCtx.getMetadataProperties().getMetadataNodeName();
-        ftStrategy = appCtx.getFaultToleranceStrategy();
-        ftStrategy.bindTo(this);
+        lifecycleCoordinator = appCtx.getNcLifecycleCoordinator();
+        lifecycleCoordinator.bindTo(this);
     }
 
     @Override
@@ -98,7 +92,7 @@ public class ClusterStateManager implements IClusterStateManager {
         failedNodes.add(nodeId);
         ncConfigMap.remove(nodeId);
         pendingRemoval.remove(nodeId);
-        ftStrategy.notifyNodeFailure(nodeId);
+        lifecycleCoordinator.notifyNodeFailure(nodeId);
     }
 
     @Override
@@ -109,7 +103,7 @@ public class ClusterStateManager implements IClusterStateManager {
         failedNodes.remove(nodeId);
         ncConfigMap.put(nodeId, configuration);
         updateNodeConfig(nodeId, configuration);
-        ftStrategy.notifyNodeJoin(nodeId);
+        lifecycleCoordinator.notifyNodeJoin(nodeId);
     }
 
     @Override


Mime
View raw message