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/02/19 20:30:30 UTC
[4/6] asterixdb git commit: Introduce Strategy Based Replication and
Fault-Tolerance
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/ef173f34/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
index bc270df..ce57648 100644
--- 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
@@ -18,27 +18,17 @@
*/
package org.apache.asterix.hyracks.bootstrap;
-import java.io.File;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.logging.Level;
-import java.util.logging.Logger;
-
-import org.apache.asterix.app.external.ExternalLibraryUtils;
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.MetadataProperties;
-import org.apache.asterix.common.config.StorageProperties;
-import org.apache.asterix.common.config.TransactionProperties;
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.replication.IRemoteRecoveryManager;
-import org.apache.asterix.common.transactions.ICheckpointManager;
+import org.apache.asterix.common.config.MetadataProperties;
+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;
@@ -47,39 +37,39 @@ 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.runtime.message.ReportMaxResourceIdMessage;
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.application.INCApplicationEntryPoint;
import org.apache.hyracks.api.job.resource.NodeCapacity;
-import org.apache.hyracks.api.lifecycle.ILifeCycleComponentManager;
-import org.apache.hyracks.api.lifecycle.LifeCycleComponentManager;
import org.apache.hyracks.api.messages.IMessageBroker;
import org.apache.hyracks.control.nc.NodeControllerService;
import org.kohsuke.args4j.CmdLineException;
import org.kohsuke.args4j.CmdLineParser;
import org.kohsuke.args4j.Option;
+import java.io.File;
+import java.util.Collections;
+import java.util.List;
+import java.util.logging.Level;
+import java.util.logging.Logger;
+
public class NCApplicationEntryPoint implements INCApplicationEntryPoint {
private static final Logger LOGGER = Logger.getLogger(NCApplicationEntryPoint.class.getName());
- @Option(name = "-initial-run",
- usage = "A flag indicating if it's the first time the NC is started (default: false)", required = false)
+ @Option(name = "-initial-run", usage = "A flag indicating if it's the first time the NC is started "
+ + "(default: false)", required = false)
public boolean initialRun = false;
- @Option(name = "-virtual-NC",
- usage = "A flag indicating if this NC is running on virtual cluster " + "(default: false)",
- required = false)
+ @Option(name = "-virtual-NC", usage = "A flag indicating if this NC is running on virtual cluster "
+ + "(default: false)", required = false)
public boolean virtualNC = false;
private INCApplicationContext ncApplicationContext = null;
private IAppRuntimeContext runtimeContext;
private String nodeId;
- private boolean isMetadataNode = false;
private boolean stopInitiated = false;
- private SystemState systemState = SystemState.NEW_UNIVERSE;
- private boolean pendingFailbackCompletion = false;
+ private SystemState systemState;
private IMessageBroker messageBroker;
@Override
@@ -93,8 +83,8 @@ public class NCApplicationEntryPoint implements INCApplicationEntryPoint {
parser.printUsage(System.err);
throw e;
}
- ncAppCtx.setThreadFactory(new AsterixThreadFactory(ncAppCtx.getThreadFactory(),
- ncAppCtx.getLifeCycleComponentManager()));
+ ncAppCtx.setThreadFactory(
+ new AsterixThreadFactory(ncAppCtx.getThreadFactory(), ncAppCtx.getLifeCycleComponentManager()));
ncApplicationContext = ncAppCtx;
nodeId = ncApplicationContext.getNodeId();
if (LOGGER.isLoggable(Level.INFO)) {
@@ -104,12 +94,11 @@ public class NCApplicationEntryPoint implements INCApplicationEntryPoint {
final NodeControllerService controllerService = (NodeControllerService) ncAppCtx.getControllerService();
if (System.getProperty("java.rmi.server.hostname") == null) {
- System.setProperty("java.rmi.server.hostname", (controllerService)
- .getConfiguration().clusterNetPublicIPAddress);
+ System.setProperty("java.rmi.server.hostname",
+ (controllerService).getConfiguration().clusterNetPublicIPAddress);
}
runtimeContext = new NCAppRuntimeContext(ncApplicationContext, getExtensions());
- MetadataProperties metadataProperties = ((IPropertiesProvider) runtimeContext)
- .getMetadataProperties();
+ MetadataProperties metadataProperties = ((IPropertiesProvider) runtimeContext).getMetadataProperties();
if (!metadataProperties.getNodeNames().contains(ncApplicationContext.getNodeId())) {
if (LOGGER.isLoggable(Level.INFO)) {
LOGGER.info("Substitute node joining : " + ncApplicationContext.getNodeId());
@@ -118,69 +107,35 @@ public class NCApplicationEntryPoint implements INCApplicationEntryPoint {
}
runtimeContext.initialize(initialRun);
ncApplicationContext.setApplicationObject(runtimeContext);
- MessagingProperties messagingProperties = ((IPropertiesProvider) runtimeContext)
- .getMessagingProperties();
+ MessagingProperties messagingProperties = ((IPropertiesProvider) runtimeContext).getMessagingProperties();
messageBroker = new NCMessageBroker(controllerService, messagingProperties);
ncApplicationContext.setMessageBroker(messageBroker);
MessagingChannelInterfaceFactory interfaceFactory = new MessagingChannelInterfaceFactory(
(NCMessageBroker) messageBroker, messagingProperties);
ncApplicationContext.setMessagingChannelInterfaceFactory(interfaceFactory);
- boolean replicationEnabled = ClusterProperties.INSTANCE.isReplicationEnabled();
- boolean autoFailover = ClusterProperties.INSTANCE.isAutoFailoverEnabled();
- if (initialRun) {
- LOGGER.info("System is being initialized. (first run)");
- } else {
- IRecoveryManager recoveryMgr = runtimeContext.getTransactionSubsystem().getRecoveryManager();
- systemState = recoveryMgr.getSystemState();
+ IRecoveryManager recoveryMgr = runtimeContext.getTransactionSubsystem().getRecoveryManager();
+ systemState = recoveryMgr.getSystemState();
+ if (systemState == SystemState.NEW_UNIVERSE) {
if (LOGGER.isLoggable(Level.INFO)) {
- LOGGER.info("System is in a state: " + systemState);
- }
-
- //do not attempt to perform remote recovery if this is a virtual NC
- if (autoFailover && !virtualNC) {
- if (systemState == SystemState.NEW_UNIVERSE || systemState == SystemState.CORRUPTED) {
- //Start failback process
- IRemoteRecoveryManager remoteRecoveryMgr = runtimeContext.getRemoteRecoveryManager();
- remoteRecoveryMgr.startFailbackProcess();
- systemState = SystemState.RECOVERING;
- pendingFailbackCompletion = true;
- }
- } else {
- //recover if the system is corrupted by checking system state.
- if (systemState == SystemState.CORRUPTED) {
- recoveryMgr.startRecovery(true);
- }
+ LOGGER.info("System state: " + SystemState.NEW_UNIVERSE);
+ 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());
}
- /**
- * if the node pending failback completion, the replication channel
- * should not be opened to avoid other nodes connecting to it before
- * the node completes its failback. CC will notify other replicas once
- * this node is ready to receive replication requests.
- */
- if (replicationEnabled && !pendingFailbackCompletion) {
- startReplicationService();
- }
+ performLocalCleanUp();
}
protected List<AsterixExtension> getExtensions() {
return Collections.emptyList();
}
- private void startReplicationService() throws InterruptedException {
- //Open replication channel
- runtimeContext.getReplicationChannel().start();
-
- //Check the state of remote replicas
- runtimeContext.getReplicationManager().initializeReplicasState();
-
- //Start replication after the state of remote replicas has been initialized.
- runtimeContext.getReplicationManager().startReplicationThreads();
- }
-
@Override
public void stop() throws Exception {
if (!stopInitiated) {
@@ -204,63 +159,13 @@ public class NCApplicationEntryPoint implements INCApplicationEntryPoint {
@Override
public void notifyStartupComplete() throws Exception {
- //Send max resource id on this NC to the CC
- ReportMaxResourceIdMessage.send((NodeControllerService) ncApplicationContext.getControllerService());
- MetadataProperties metadataProperties = ((IPropertiesProvider) runtimeContext)
- .getMetadataProperties();
- if (initialRun || systemState == SystemState.NEW_UNIVERSE) {
- if (LOGGER.isLoggable(Level.INFO)) {
- LOGGER.info("System state: " + SystemState.NEW_UNIVERSE);
- 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());
- }
-
- isMetadataNode = nodeId.equals(metadataProperties.getMetadataNodeName());
- if (isMetadataNode && !pendingFailbackCompletion) {
- runtimeContext.initializeMetadata(systemState == SystemState.NEW_UNIVERSE);
+ // Since we don't pass initial run flag in AsterixHyracksIntegrationUtil, we use the virtualNC flag
+ if (systemState == SystemState.NEW_UNIVERSE && (initialRun || virtualNC)) {
+ systemState = SystemState.INITIAL_RUN;
}
- ExternalLibraryUtils.setUpExternaLibraries(runtimeContext.getLibraryManager(),
- isMetadataNode && !pendingFailbackCompletion);
-
- if (LOGGER.isLoggable(Level.INFO)) {
- LOGGER.info("Starting lifecycle components");
- }
-
- Map<String, String> lifecycleMgmtConfiguration = new HashMap<String, String>();
- String dumpPathKey = LifeCycleComponentManager.Config.DUMP_PATH_KEY;
- String dumpPath = metadataProperties.getCoredumpPath(nodeId);
- lifecycleMgmtConfiguration.put(dumpPathKey, dumpPath);
- if (LOGGER.isLoggable(Level.INFO)) {
- LOGGER.info("Coredump directory for NC is: " + dumpPath);
- }
- ILifeCycleComponentManager lccm = ncApplicationContext.getLifeCycleComponentManager();
- lccm.configure(lifecycleMgmtConfiguration);
- if (LOGGER.isLoggable(Level.INFO)) {
- LOGGER.info("Configured:" + lccm);
- }
- ncApplicationContext.setStateDumpHandler(
- new AsterixStateDumpHandler(ncApplicationContext.getNodeId(), lccm.getDumpPath(), lccm));
-
- lccm.startAll();
-
- if (!pendingFailbackCompletion) {
- ICheckpointManager checkpointMgr = runtimeContext.getTransactionSubsystem().getCheckpointManager();
- checkpointMgr.doSharpCheckpoint();
-
- if (isMetadataNode) {
- runtimeContext.exportMetadataNodeStub();
- }
- }
-
- //Clean any temporary files
- performLocalCleanUp();
+ // Request startup tasks from CC
+ StartupTaskRequestMessage.send((NodeControllerService) ncApplicationContext.getControllerService(),
+ systemState);
}
@Override
@@ -296,8 +201,7 @@ public class NCApplicationEntryPoint implements INCApplicationEntryPoint {
}
private void updateOnNodeJoin() {
- MetadataProperties metadataProperties = ((IPropertiesProvider) runtimeContext)
- .getMetadataProperties();
+ MetadataProperties metadataProperties = ((IPropertiesProvider) runtimeContext).getMetadataProperties();
if (!metadataProperties.getNodeNames().contains(nodeId)) {
metadataProperties.getNodeNames().add(nodeId);
Cluster cluster = ClusterProperties.INSTANCE.getCluster();
@@ -305,8 +209,7 @@ public class NCApplicationEntryPoint implements INCApplicationEntryPoint {
throw new IllegalStateException("No cluster configuration found for this instance");
}
String asterixInstanceName = metadataProperties.getInstanceName();
- TransactionProperties txnProperties = ((IPropertiesProvider) runtimeContext)
- .getTransactionProperties();
+ TransactionProperties txnProperties = ((IPropertiesProvider) runtimeContext).getTransactionProperties();
Node self = null;
List<Node> nodes;
if (cluster.getSubstituteNodes() != null) {
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/ef173f34/asterixdb/asterix-app/src/main/java/org/apache/asterix/util/FaultToleranceUtil.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-app/src/main/java/org/apache/asterix/util/FaultToleranceUtil.java b/asterixdb/asterix-app/src/main/java/org/apache/asterix/util/FaultToleranceUtil.java
new file mode 100644
index 0000000..0a9a215
--- /dev/null
+++ b/asterixdb/asterix-app/src/main/java/org/apache/asterix/util/FaultToleranceUtil.java
@@ -0,0 +1,67 @@
+/*
+ * 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.util;
+
+import java.util.List;
+import java.util.Map;
+import java.util.logging.Level;
+import java.util.logging.Logger;
+import java.util.stream.Collectors;
+
+import org.apache.asterix.common.cluster.IClusterStateManager;
+import org.apache.asterix.common.messaging.api.ICCMessageBroker;
+import org.apache.asterix.common.replication.IReplicationStrategy;
+import org.apache.asterix.common.replication.Replica;
+import org.apache.asterix.runtime.message.ReplicaEventMessage;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.hyracks.api.application.IClusterLifecycleListener.ClusterEventType;
+
+public class FaultToleranceUtil {
+
+ private static final Logger LOGGER = Logger.getLogger(FaultToleranceUtil.class.getName());
+ private static final String CLUSTER_NET_IP_ADDRESS_KEY = "cluster-net-ip-address";
+
+ private FaultToleranceUtil() {
+ throw new AssertionError();
+ }
+
+ public static void notifyImpactedReplicas(String nodeId, ClusterEventType event,
+ IClusterStateManager clusterManager, ICCMessageBroker messageBroker,
+ IReplicationStrategy replicationStrategy) {
+ List<String> primaryRemoteReplicas = replicationStrategy.getRemotePrimaryReplicas(nodeId).stream()
+ .map(Replica::getId).collect(Collectors.toList());
+ String nodeIdAddress = StringUtils.EMPTY;
+ Map<String, Map<String, String>> activeNcConfiguration = clusterManager.getActiveNcConfiguration();
+ // In case the node joined with a new IP address, we need to send it to the other replicas
+ if (event == ClusterEventType.NODE_JOIN) {
+ nodeIdAddress = activeNcConfiguration.get(nodeId).get(CLUSTER_NET_IP_ADDRESS_KEY);
+ }
+ ReplicaEventMessage msg = new ReplicaEventMessage(nodeId, nodeIdAddress, event);
+ for (String replica : primaryRemoteReplicas) {
+ // If the remote replica is alive, send the event
+ if (activeNcConfiguration.containsKey(replica)) {
+ try {
+ messageBroker.sendApplicationMessageToNC(msg, replica);
+ } catch (Exception e) {
+ LOGGER.log(Level.WARNING, "Failed sending an application message to an NC", e);
+ }
+ }
+ }
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/ef173f34/asterixdb/asterix-app/src/main/resources/cluster.xml
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-app/src/main/resources/cluster.xml b/asterixdb/asterix-app/src/main/resources/cluster.xml
index 8f0b694..7f78b26 100644
--- a/asterixdb/asterix-app/src/main/resources/cluster.xml
+++ b/asterixdb/asterix-app/src/main/resources/cluster.xml
@@ -19,14 +19,22 @@
<cluster xmlns="cluster">
<instance_name>asterix</instance_name>
<store>storage</store>
+ <metadata_node>nc1</metadata_node>
- <data_replication>
+ <high_availability>
<enabled>false</enabled>
- <replication_port>2016</replication_port>
- <replication_factor>2</replication_factor>
- <auto_failover>false</auto_failover>
- <replication_time_out>30</replication_time_out>
- </data_replication>
+ <data_replication>
+ <strategy>metadata_only</strategy>
+ <replication_port>2016</replication_port>
+ <replication_time_out>30</replication_time_out>
+ </data_replication>
+ <fault_tolerance>
+ <strategy>metadata_node</strategy>
+ <replica>
+ <node_id>nc2</node_id>
+ </replica>
+ </fault_tolerance>
+ </high_availability>
<master_node>
<id>master</id>
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/ef173f34/asterixdb/asterix-app/src/test/resources/runtimets/results/api/replication/replication.1.adm
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/results/api/replication/replication.1.adm b/asterixdb/asterix-app/src/test/resources/runtimets/results/api/replication/replication.1.adm
index 74e019b..a614faa 100644
--- a/asterixdb/asterix-app/src/test/resources/runtimets/results/api/replication/replication.1.adm
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/results/api/replication/replication.1.adm
@@ -1,11 +1,9 @@
{
"config" : {
- "enabled" : false,
- "factor" : 2,
"log.batchsize" : 4096,
"log.buffer.numpages" : 8,
"log.buffer.pagesize" : 131072,
"max.remote.recovery.attempts" : 5,
"timeout" : 30
}
-}
\ No newline at end of file
+}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/ef173f34/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/api/IDatasetLifecycleManager.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/api/IDatasetLifecycleManager.java b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/api/IDatasetLifecycleManager.java
index 04e30ff..ae19f23 100644
--- a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/api/IDatasetLifecycleManager.java
+++ b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/api/IDatasetLifecycleManager.java
@@ -23,6 +23,7 @@ import java.util.List;
import org.apache.asterix.common.context.DatasetInfo;
import org.apache.asterix.common.context.IndexInfo;
import org.apache.asterix.common.context.PrimaryIndexOperationTracker;
+import org.apache.asterix.common.replication.IReplicationStrategy;
import org.apache.hyracks.api.exceptions.HyracksDataException;
import org.apache.hyracks.storage.am.common.api.IIndex;
import org.apache.hyracks.storage.am.common.api.IResourceLifecycleManager;
@@ -102,4 +103,12 @@ public interface IDatasetLifecycleManager extends IResourceLifecycleManager<IInd
* @throws HyracksDataException
*/
void closeUserDatasets() throws HyracksDataException;
+
+ /**
+ * Flushes all opened datasets that are matching {@code replicationStrategy}.
+ *
+ * @param replicationStrategy
+ * @throws HyracksDataException
+ */
+ void flushDataset(IReplicationStrategy replicationStrategy) throws HyracksDataException;
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/ef173f34/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/api/INCLifecycleTask.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/api/INCLifecycleTask.java b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/api/INCLifecycleTask.java
new file mode 100644
index 0000000..c30e999
--- /dev/null
+++ b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/api/INCLifecycleTask.java
@@ -0,0 +1,36 @@
+/*
+ * 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.api;
+
+import java.io.Serializable;
+
+import org.apache.hyracks.api.exceptions.HyracksDataException;
+import org.apache.hyracks.api.service.IControllerService;
+
+@FunctionalInterface
+public interface INCLifecycleTask extends Serializable {
+
+ /**
+ * Performs the task.
+ *
+ * @param cs
+ * @throws HyracksDataException
+ */
+ void perform(IControllerService cs) throws HyracksDataException;
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/ef173f34/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/cluster/IClusterStateManager.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/cluster/IClusterStateManager.java b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/cluster/IClusterStateManager.java
new file mode 100644
index 0000000..d971f48
--- /dev/null
+++ b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/cluster/IClusterStateManager.java
@@ -0,0 +1,85 @@
+/*
+ * 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.cluster;
+
+import java.util.Map;
+
+import org.apache.asterix.common.api.IClusterManagementWork.ClusterState;
+import org.apache.hyracks.api.exceptions.HyracksDataException;
+
+public interface IClusterStateManager {
+
+ /**
+ * @return The current cluster state.
+ */
+ ClusterState getState();
+
+ /**
+ * Updates the cluster state based on the state of all cluster partitions and the metadata node.
+ * Cluster state after refresh:
+ * ACTIVE: all cluster partitions are active and the metadata node is bound.
+ * PENDING: all cluster partitions are active but the metadata node is not bound.
+ * UNUSABLE: one or more cluster partitions are not active.
+ */
+ void refreshState() throws HyracksDataException;
+
+ /**
+ * Sets the cluster state into {@code state}
+ */
+ void setState(ClusterState state);
+
+ /**
+ * Updates all partitions of {@code nodeId} based on the {@code active} flag.
+ * @param nodeId
+ * @param active
+ * @throws HyracksDataException
+ */
+ void updateNodePartitions(String nodeId, boolean active) throws HyracksDataException;
+
+ /**
+ * Updates the active node and active state of the cluster partition with id {@code partitionNum}
+ */
+ void updateClusterPartition(Integer partitionNum, String activeNode, boolean active);
+
+ /**
+ * Updates the metadata node id and its state.
+ */
+ void updateMetadataNode(String nodeId, boolean active);
+
+ /**
+ * @return a map of nodeId and NC Configuration for active nodes.
+ */
+ Map<String, Map<String, String>> getActiveNcConfiguration();
+
+ /**
+ * @return The current metadata node Id.
+ */
+ String getCurrentMetadataNodeId();
+
+ /**
+ * @param nodeId
+ * @return The node originally assigned partitions.
+ */
+ ClusterPartition[] getNodePartitions(String nodeId);
+
+ /**
+ * @return A copy of the current state of the cluster partitions.
+ */
+ ClusterPartition[] getClusterPartitons();
+}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/ef173f34/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/config/ClusterProperties.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/config/ClusterProperties.java b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/config/ClusterProperties.java
index 81c5a6d..980ad24 100644
--- a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/config/ClusterProperties.java
+++ b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/config/ClusterProperties.java
@@ -19,21 +19,27 @@
package org.apache.asterix.common.config;
import java.io.InputStream;
+import java.util.Optional;
+import java.util.stream.Collectors;
import javax.xml.bind.JAXBContext;
import javax.xml.bind.JAXBException;
import javax.xml.bind.Unmarshaller;
+import org.apache.asterix.common.replication.IReplicationStrategy;
+import org.apache.asterix.common.replication.ReplicationStrategyFactory;
import org.apache.asterix.event.schema.cluster.Cluster;
+import org.apache.asterix.event.schema.cluster.Node;
+import org.apache.asterix.event.schema.cluster.Replica;
+import org.apache.commons.lang3.StringUtils;
public class ClusterProperties {
public static final ClusterProperties INSTANCE = new ClusterProperties();
-
private static final String CLUSTER_CONFIGURATION_FILE = "cluster.xml";
private static final String DEFAULT_STORAGE_DIR_NAME = "storage";
-
- private final Cluster cluster;
+ private String nodeNamePrefix = StringUtils.EMPTY;
+ private Cluster cluster;
private ClusterProperties() {
InputStream is = this.getClass().getClassLoader().getResourceAsStream(CLUSTER_CONFIGURATION_FILE);
@@ -42,11 +48,11 @@ public class ClusterProperties {
JAXBContext ctx = JAXBContext.newInstance(Cluster.class);
Unmarshaller unmarshaller = ctx.createUnmarshaller();
cluster = (Cluster) unmarshaller.unmarshal(is);
+ nodeNamePrefix = cluster.getInstanceName() + "_";
+ updateNodeIdToFullName();
} catch (JAXBException e) {
throw new IllegalStateException("Failed to read configuration file " + CLUSTER_CONFIGURATION_FILE, e);
}
- } else {
- cluster = null;
}
}
@@ -62,14 +68,41 @@ public class ClusterProperties {
return DEFAULT_STORAGE_DIR_NAME;
}
- public boolean isReplicationEnabled() {
- if (cluster != null && cluster.getDataReplication() != null) {
- return cluster.getDataReplication().isEnabled();
+ public Node getNodeById(String nodeId) {
+ Optional<Node> matchingNode = cluster.getNode().stream().filter(node -> node.getId().equals(nodeId)).findAny();
+ return matchingNode.isPresent() ? matchingNode.get() : null;
+ }
+
+ public int getNodeIndex(String nodeId) {
+ for (int i = 0; i < cluster.getNode().size(); i++) {
+ Node node = cluster.getNode().get(i);
+ if (node.getId().equals(nodeId)) {
+ return i;
+ }
+ }
+ return -1;
+ }
+
+ public IReplicationStrategy getReplicationStrategy() {
+ return ReplicationStrategyFactory.create(cluster);
+ }
+
+ private String getNodeFullName(String nodeId) {
+ if (nodeId.startsWith(nodeNamePrefix)) {
+ return nodeId;
}
- return false;
+ return nodeNamePrefix + nodeId;
}
- public boolean isAutoFailoverEnabled() {
- return isReplicationEnabled() && cluster.getDataReplication().isAutoFailover();
+ private void updateNodeIdToFullName() {
+ cluster.getNode().forEach(node -> node.setId(getNodeFullName(node.getId())));
+ if (cluster.getMetadataNode() != null) {
+ cluster.setMetadataNode(getNodeFullName(cluster.getMetadataNode()));
+ }
+ if (cluster.getHighAvailability() != null && cluster.getHighAvailability().getFaultTolerance() != null
+ && cluster.getHighAvailability().getFaultTolerance().getReplica() != null) {
+ Replica replicas = cluster.getHighAvailability().getFaultTolerance().getReplica();
+ replicas.setNodeId(replicas.getNodeId().stream().map(this::getNodeFullName).collect(Collectors.toList()));
+ }
}
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/ef173f34/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/config/ReplicationProperties.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/config/ReplicationProperties.java b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/config/ReplicationProperties.java
index 164a525..cf2ce4f 100644
--- a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/config/ReplicationProperties.java
+++ b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/config/ReplicationProperties.java
@@ -18,11 +18,10 @@
*/
package org.apache.asterix.common.config;
-import java.util.HashSet;
import java.util.Set;
-import java.util.logging.Level;
-import java.util.logging.Logger;
+import java.util.stream.Collectors;
+import org.apache.asterix.common.replication.IReplicationStrategy;
import org.apache.asterix.common.replication.Replica;
import org.apache.asterix.event.schema.cluster.Cluster;
import org.apache.asterix.event.schema.cluster.Node;
@@ -31,16 +30,8 @@ import org.apache.hyracks.util.StorageUtil.StorageUnit;
public class ReplicationProperties extends AbstractProperties {
- private static final Logger LOGGER = Logger.getLogger(ReplicationProperties.class.getName());
-
-
private static final int REPLICATION_DATAPORT_DEFAULT = 2000;
- private static final String REPLICATION_ENABLED_KEY = "replication.enabled";
-
- private static final String REPLICATION_FACTOR_KEY = "replication.factor";
- private static final int REPLICATION_FACTOR_DEFAULT = 1;
-
private static final String REPLICATION_TIMEOUT_KEY = "replication.timeout";
private static final int REPLICATION_TIME_OUT_DEFAULT = 15;
@@ -60,205 +51,60 @@ public class ReplicationProperties extends AbstractProperties {
private static final int REPLICATION_LOG_BUFFER_PAGE_SIZE_DEFAULT = StorageUtil.getSizeInBytes(128,
StorageUnit.KILOBYTE);
- private final String nodeNamePrefix;
private final Cluster cluster;
+ private final IReplicationStrategy repStrategy;
public ReplicationProperties(PropertiesAccessor accessor) {
super(accessor);
this.cluster = ClusterProperties.INSTANCE.getCluster();
-
- if (cluster != null) {
- nodeNamePrefix = cluster.getInstanceName() + "_";
- } else {
- nodeNamePrefix = "";
- }
- }
-
- @PropertyKey(REPLICATION_ENABLED_KEY)
- public boolean isReplicationEnabled() {
- return ClusterProperties.INSTANCE.isReplicationEnabled();
+ this.repStrategy = ClusterProperties.INSTANCE.getReplicationStrategy();
}
public String getReplicaIPAddress(String nodeId) {
- if (cluster != null) {
- for (int i = 0; i < cluster.getNode().size(); i++) {
- Node node = cluster.getNode().get(i);
- if (getRealCluserNodeID(node.getId()).equals(nodeId)) {
- return node.getClusterIp();
- }
- }
- }
- return NODE_IP_ADDRESS_DEFAULT;
+ Node node = ClusterProperties.INSTANCE.getNodeById(nodeId);
+ return node != null ? node.getClusterIp() : NODE_IP_ADDRESS_DEFAULT;
}
public int getDataReplicationPort(String nodeId) {
- if (cluster != null && cluster.getDataReplication() != null) {
- for (int i = 0; i < cluster.getNode().size(); i++) {
- Node node = cluster.getNode().get(i);
- if (getRealCluserNodeID(node.getId()).equals(nodeId)) {
- return node.getReplicationPort() != null ? node.getReplicationPort().intValue()
- : cluster.getDataReplication().getReplicationPort().intValue();
- }
- }
+ Node node = ClusterProperties.INSTANCE.getNodeById(nodeId);
+ if (node != null) {
+ return node.getReplicationPort() != null ? node.getReplicationPort().intValue()
+ : cluster.getHighAvailability().getDataReplication().getReplicationPort().intValue();
}
return REPLICATION_DATAPORT_DEFAULT;
}
- public Set<Replica> getRemoteReplicas(String nodeId) {
- Set<Replica> remoteReplicas = new HashSet<>();;
-
- int numberOfRemoteReplicas = getReplicationFactor() - 1;
- //Using chained-declustering
- if (cluster != null) {
- int nodeIndex = -1;
- //find the node index in the cluster config
- for (int i = 0; i < cluster.getNode().size(); i++) {
- Node node = cluster.getNode().get(i);
- if (getRealCluserNodeID(node.getId()).equals(nodeId)) {
- nodeIndex = i;
- break;
- }
- }
-
- if (nodeIndex == -1) {
- LOGGER.log(Level.WARNING,
- "Could not find node " + getRealCluserNodeID(nodeId) + " in cluster configurations");
- return null;
- }
-
- //find nodes to the right of this node
- for (int i = nodeIndex + 1; i < cluster.getNode().size(); i++) {
- remoteReplicas.add(getReplicaByNodeIndex(i));
- if (remoteReplicas.size() == numberOfRemoteReplicas) {
- break;
- }
- }
-
- //if not all remote replicas have been found, start from the beginning
- if (remoteReplicas.size() != numberOfRemoteReplicas) {
- for (int i = 0; i < cluster.getNode().size(); i++) {
- remoteReplicas.add(getReplicaByNodeIndex(i));
- if (remoteReplicas.size() == numberOfRemoteReplicas) {
- break;
- }
- }
- }
- }
- return remoteReplicas;
- }
-
- private Replica getReplicaByNodeIndex(int nodeIndex) {
- Node node = cluster.getNode().get(nodeIndex);
- Node replicaNode = new Node();
- replicaNode.setId(getRealCluserNodeID(node.getId()));
- replicaNode.setClusterIp(node.getClusterIp());
- return new Replica(replicaNode);
- }
-
public Replica getReplicaById(String nodeId) {
- int nodeIndex = -1;
- if (cluster != null) {
- for (int i = 0; i < cluster.getNode().size(); i++) {
- Node node = cluster.getNode().get(i);
-
- if (getRealCluserNodeID(node.getId()).equals(nodeId)) {
- nodeIndex = i;
- break;
- }
- }
+ Node node = ClusterProperties.INSTANCE.getNodeById(nodeId);
+ if (node != null) {
+ return new Replica(node);
}
-
- if (nodeIndex < 0) {
- return null;
- }
-
- return getReplicaByNodeIndex(nodeIndex);
+ return null;
}
public Set<String> getRemoteReplicasIds(String nodeId) {
- Set<String> remoteReplicasIds = new HashSet<>();
- Set<Replica> remoteReplicas = getRemoteReplicas(nodeId);
-
- for (Replica replica : remoteReplicas) {
- remoteReplicasIds.add(replica.getId());
- }
-
- return remoteReplicasIds;
+ return repStrategy.getRemoteReplicas(nodeId).stream().map(Replica::getId).collect(Collectors.toSet());
}
- public String getRealCluserNodeID(String nodeId) {
- return nodeNamePrefix + nodeId;
+ public Set<String> getRemotePrimaryReplicasIds(String nodeId) {
+ return repStrategy.getRemotePrimaryReplicas(nodeId).stream().map(Replica::getId).collect(Collectors.toSet());
}
public Set<String> getNodeReplicasIds(String nodeId) {
- Set<String> replicaIds = new HashSet<>();
- replicaIds.add(nodeId);
- replicaIds.addAll(getRemoteReplicasIds(nodeId));
- return replicaIds;
- }
-
- @PropertyKey(REPLICATION_FACTOR_KEY)
- public int getReplicationFactor() {
- if (cluster != null) {
- if (cluster.getDataReplication() == null || cluster.getDataReplication().getReplicationFactor() == null) {
- return REPLICATION_FACTOR_DEFAULT;
- }
- return cluster.getDataReplication().getReplicationFactor().intValue();
- }
- return REPLICATION_FACTOR_DEFAULT;
+ Set<String> remoteReplicasIds = getRemoteReplicasIds(nodeId);
+ // This includes the node itself
+ remoteReplicasIds.add(nodeId);
+ return remoteReplicasIds;
}
@PropertyKey(REPLICATION_TIMEOUT_KEY)
public int getReplicationTimeOut() {
if (cluster != null) {
- return cluster.getDataReplication().getReplicationTimeOut().intValue();
+ return cluster.getHighAvailability().getDataReplication().getReplicationTimeOut().intValue();
}
return REPLICATION_TIME_OUT_DEFAULT;
}
- /**
- * @param nodeId
- * @return The set of nodes which replicate to this node, including the node itself
- */
- public Set<String> getNodeReplicationClients(String nodeId) {
- Set<String> clientReplicas = new HashSet<>();
- clientReplicas.add(nodeId);
-
- int clientsCount = getReplicationFactor();
-
- //Using chained-declustering backwards
- if (cluster != null) {
- int nodeIndex = -1;
- //find the node index in the cluster config
- for (int i = 0; i < cluster.getNode().size(); i++) {
- Node node = cluster.getNode().get(i);
- if (getRealCluserNodeID(node.getId()).equals(nodeId)) {
- nodeIndex = i;
- break;
- }
- }
-
- //find nodes to the left of this node
- for (int i = nodeIndex - 1; i >= 0; i--) {
- clientReplicas.add(getReplicaByNodeIndex(i).getId());
- if (clientReplicas.size() == clientsCount) {
- break;
- }
- }
-
- //if not all client replicas have been found, start from the end
- if (clientReplicas.size() != clientsCount) {
- for (int i = cluster.getNode().size() - 1; i >= 0; i--) {
- clientReplicas.add(getReplicaByNodeIndex(i).getId());
- if (clientReplicas.size() == clientsCount) {
- break;
- }
- }
- }
- }
- return clientReplicas;
- }
-
@PropertyKey(REPLICATION_MAX_REMOTE_RECOVERY_ATTEMPTS_KEY)
public int getMaxRemoteRecoveryAttempts() {
return MAX_REMOTE_RECOVERY_ATTEMPTS;
@@ -281,4 +127,12 @@ public class ReplicationProperties extends AbstractProperties {
return accessor.getProperty(REPLICATION_LOG_BATCH_SIZE_KEY, REPLICATION_LOG_BATCH_SIZE_DEFAULT,
PropertyInterpreters.getIntegerBytePropertyInterpreter());
}
-}
+
+ public boolean isParticipant(String nodeId) {
+ return repStrategy.isParticipant(nodeId);
+ }
+
+ public IReplicationStrategy getReplicationStrategy() {
+ return repStrategy;
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/ef173f34/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/context/DatasetLifecycleManager.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/context/DatasetLifecycleManager.java b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/context/DatasetLifecycleManager.java
index f49b07a..cbb4868 100644
--- a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/context/DatasetLifecycleManager.java
+++ b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/context/DatasetLifecycleManager.java
@@ -30,6 +30,7 @@ import org.apache.asterix.common.api.IDatasetLifecycleManager;
import org.apache.asterix.common.config.StorageProperties;
import org.apache.asterix.common.exceptions.ACIDException;
import org.apache.asterix.common.ioopcallbacks.AbstractLSMIOOperationCallback;
+import org.apache.asterix.common.replication.IReplicationStrategy;
import org.apache.asterix.common.transactions.ILogManager;
import org.apache.asterix.common.transactions.LogRecord;
import org.apache.asterix.common.transactions.Resource;
@@ -571,4 +572,13 @@ public class DatasetLifecycleManager implements IDatasetLifecycleManager, ILifeC
public int getNumPartitions() {
return numPartitions;
}
+
+ @Override
+ public void flushDataset(IReplicationStrategy replicationStrategy) throws HyracksDataException {
+ for (DatasetResource dsr : datasets.values()) {
+ if (replicationStrategy.isMatch(dsr.getDatasetID())) {
+ flushDatasetOpenIndexes(dsr.getDatasetInfo(), false);
+ }
+ }
+ }
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/ef173f34/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/replication/ChainedDeclusteringReplicationStrategy.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/replication/ChainedDeclusteringReplicationStrategy.java b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/replication/ChainedDeclusteringReplicationStrategy.java
new file mode 100644
index 0000000..ad326b2
--- /dev/null
+++ b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/replication/ChainedDeclusteringReplicationStrategy.java
@@ -0,0 +1,85 @@
+/*
+ * 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 java.util.Collections;
+import java.util.HashSet;
+import java.util.Set;
+import java.util.logging.Level;
+import java.util.logging.Logger;
+
+import org.apache.asterix.common.config.ClusterProperties;
+import org.apache.asterix.event.schema.cluster.Cluster;
+
+public class ChainedDeclusteringReplicationStrategy implements IReplicationStrategy {
+
+ private static final Logger LOGGER = Logger.getLogger(ChainedDeclusteringReplicationStrategy.class.getName());
+ private int replicationFactor;
+
+ @Override
+ public boolean isMatch(int datasetId) {
+ return true;
+ }
+
+ @Override
+ public Set<Replica> getRemoteReplicas(String nodeId) {
+ Set<Replica> remoteReplicas = new HashSet<>();
+ Cluster cluster = ClusterProperties.INSTANCE.getCluster();
+ int numberOfRemoteReplicas = replicationFactor - 1;
+ int nodeIndex = ClusterProperties.INSTANCE.getNodeIndex(nodeId);
+
+ if (nodeIndex == -1) {
+ LOGGER.log(Level.WARNING, "Could not find node " + nodeId + " in cluster configurations");
+ return Collections.emptySet();
+ }
+
+ //find nodes to the right of this node
+ while (remoteReplicas.size() != numberOfRemoteReplicas) {
+ remoteReplicas.add(new Replica(cluster.getNode().get(++nodeIndex % cluster.getNode().size())));
+ }
+
+ return remoteReplicas;
+ }
+
+ @Override
+ public Set<Replica> getRemotePrimaryReplicas(String nodeId) {
+ Set<Replica> clientReplicas = new HashSet<>();
+ Cluster cluster = ClusterProperties.INSTANCE.getCluster();
+ final int remotePrimaryReplicasCount = replicationFactor - 1;
+
+ int nodeIndex = ClusterProperties.INSTANCE.getNodeIndex(nodeId);
+
+ //find nodes to the left of this node
+ while (clientReplicas.size() != remotePrimaryReplicasCount) {
+ clientReplicas.add(new Replica(cluster.getNode().get(Math.abs(--nodeIndex % cluster.getNode().size()))));
+ }
+
+ return clientReplicas;
+ }
+
+ @Override
+ public ChainedDeclusteringReplicationStrategy from(Cluster cluster) {
+ if (cluster.getHighAvailability().getDataReplication().getReplicationFactor() == null) {
+ throw new IllegalStateException("Replication factor must be specified.");
+ }
+ ChainedDeclusteringReplicationStrategy cd = new ChainedDeclusteringReplicationStrategy();
+ cd.replicationFactor = cluster.getHighAvailability().getDataReplication().getReplicationFactor().intValue();
+ return cd;
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/ef173f34/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
new file mode 100644
index 0000000..46d5d98
--- /dev/null
+++ b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/replication/IFaultToleranceStrategy.java
@@ -0,0 +1,67 @@
+/*
+ * 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.asterix.common.messaging.api.ICCMessageBroker;
+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 replicationStrategy
+ * @param messageBroker
+ * @return
+ */
+ IFaultToleranceStrategy from(IReplicationStrategy replicationStrategy, ICCMessageBroker messageBroker);
+
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/ef173f34/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/replication/INCLifecycleMessage.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/replication/INCLifecycleMessage.java b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/replication/INCLifecycleMessage.java
new file mode 100644
index 0000000..c19b0aa
--- /dev/null
+++ b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/replication/INCLifecycleMessage.java
@@ -0,0 +1,45 @@
+/*
+ * 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.messaging.api.IApplicationMessage;
+
+public interface INCLifecycleMessage extends IApplicationMessage {
+
+ public enum MessageType {
+ REPLAY_LOGS_REQUEST,
+ REPLAY_LOGS_RESPONSE,
+ PREPARE_FAILBACK_REQUEST,
+ PREPARE_FAILBACK_RESPONSE,
+ COMPLETE_FAILBACK_REQUEST,
+ COMPLETE_FAILBACK_RESPONSE,
+ STARTUP_TASK_REQUEST,
+ STARTUP_TASK_RESPONSE,
+ STARTUP_TASK_RESULT,
+ TAKEOVER_PARTITION_REQUEST,
+ TAKEOVER_PARTITION_RESPONSE,
+ TAKEOVER_METADATA_NODE_REQUEST,
+ TAKEOVER_METADATA_NODE_RESPONSE
+ }
+
+ /**
+ * @return The message type.
+ */
+ MessageType getType();
+}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/ef173f34/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/replication/IRemoteRecoveryManager.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/replication/IRemoteRecoveryManager.java b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/replication/IRemoteRecoveryManager.java
index 9f9d74b..51b826b 100644
--- a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/replication/IRemoteRecoveryManager.java
+++ b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/replication/IRemoteRecoveryManager.java
@@ -19,8 +19,11 @@
package org.apache.asterix.common.replication;
import java.io.IOException;
+import java.util.Map;
+import java.util.Set;
import org.apache.asterix.common.exceptions.ACIDException;
+import org.apache.hyracks.api.exceptions.HyracksDataException;
public interface IRemoteRecoveryManager {
@@ -46,4 +49,23 @@ public interface IRemoteRecoveryManager {
* @throws InterruptedException
*/
public void completeFailbackProcess() throws IOException, InterruptedException;
+
+ /**
+ * Replays all committed jobs logs for {@code partitions}. Optionally, flushes all datasets
+ * to convert the replayed logs into LSM Components.
+ *
+ * @param partitions
+ * @param flush
+ * @throws HyracksDataException
+ */
+ void replayReplicaPartitionLogs(Set<Integer> partitions, boolean flush) throws HyracksDataException;
+
+ /**
+ * Performs the remote recovery plan by requesting data from each specified node
+ * for each partitions specified.
+ *
+ * @param recoveryPlan
+ * @throws HyracksDataException
+ */
+ void doRemoteRecoveryPlan(Map<String, Set<Integer>> recoveryPlan) throws HyracksDataException;
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/ef173f34/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/replication/IReplicationManager.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/replication/IReplicationManager.java b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/replication/IReplicationManager.java
index 6bd1505..b969bef 100644
--- a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/replication/IReplicationManager.java
+++ b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/replication/IReplicationManager.java
@@ -50,13 +50,13 @@ public interface IReplicationManager extends IIOReplicationManager {
*
* @param remoteReplicaId
* The replica id to send the request to.
- * @param replicasDataToRecover
- * Get files that belong to those replicas.
+ * @param partitionsToRecover
+ * Get files that belong to those partitions.
* @param existingFiles
* a list of already existing files on the requester
* @throws IOException
*/
- public void requestReplicaFiles(String remoteReplicaId, Set<String> replicasDataToRecover,
+ public void requestReplicaFiles(String remoteReplicaId, Set<Integer> partitionsToRecover,
Set<String> existingFiles) throws IOException;
/**
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/ef173f34/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/replication/IReplicationStrategy.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/replication/IReplicationStrategy.java b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/replication/IReplicationStrategy.java
new file mode 100644
index 0000000..f65f6ac
--- /dev/null
+++ b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/replication/IReplicationStrategy.java
@@ -0,0 +1,58 @@
+/*
+ * 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 java.util.Set;
+
+import org.apache.asterix.event.schema.cluster.Cluster;
+
+public interface IReplicationStrategy {
+
+ /**
+ * @param datasetId
+ * @return True, if the dataset should be replicated. Otherwise false.
+ */
+ boolean isMatch(int datasetId);
+
+ /**
+ * @param nodeId
+ * @return The set of nodes that replicate data on {@code nodeId}.
+ */
+ Set<Replica> getRemotePrimaryReplicas(String nodeId);
+
+ /**
+ * @param node
+ * @return The set of nodes that {@code nodeId} replicates data to.
+ */
+ Set<Replica> getRemoteReplicas(String node);
+
+ /**
+ * @param nodeId
+ * @return true if {@code nodeId} has any remote primary replica or remote replica. Otherwise false.
+ */
+ default boolean isParticipant(String nodeId) {
+ return !getRemoteReplicas(nodeId).isEmpty() || !getRemotePrimaryReplicas(nodeId).isEmpty();
+ }
+
+ /**
+ * @param cluster
+ * @return A replication strategy based on the passed configurations.
+ */
+ IReplicationStrategy from(Cluster cluster);
+}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/ef173f34/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/replication/MetadataOnlyReplicationStrategy.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/replication/MetadataOnlyReplicationStrategy.java b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/replication/MetadataOnlyReplicationStrategy.java
new file mode 100644
index 0000000..711f06d
--- /dev/null
+++ b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/replication/MetadataOnlyReplicationStrategy.java
@@ -0,0 +1,90 @@
+/*
+ * 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 java.util.Arrays;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.Set;
+
+import org.apache.asterix.common.config.ClusterProperties;
+import org.apache.asterix.common.metadata.MetadataIndexImmutableProperties;
+import org.apache.asterix.event.schema.cluster.Cluster;
+import org.apache.asterix.event.schema.cluster.Node;
+
+public class MetadataOnlyReplicationStrategy implements IReplicationStrategy {
+
+ private String metadataNodeId;
+ private Replica metadataPrimaryReplica;
+ private Set<Replica> metadataNodeReplicas;
+
+ @Override
+ public boolean isMatch(int datasetId) {
+ return datasetId < MetadataIndexImmutableProperties.FIRST_AVAILABLE_USER_DATASET_ID && datasetId >= 0;
+ }
+
+ @Override
+ public Set<Replica> getRemoteReplicas(String nodeId) {
+ if (nodeId.equals(metadataNodeId)) {
+ return metadataNodeReplicas;
+ }
+ return Collections.emptySet();
+ }
+
+ @Override
+ public Set<Replica> getRemotePrimaryReplicas(String nodeId) {
+ if (metadataNodeReplicas.stream().map(Replica::getId).filter(replicaId -> replicaId.equals(nodeId))
+ .count() != 0) {
+ return new HashSet<>(Arrays.asList(metadataPrimaryReplica));
+ }
+ return Collections.emptySet();
+ }
+
+ @Override
+ public MetadataOnlyReplicationStrategy from(Cluster cluster) {
+ if (cluster.getMetadataNode() == null) {
+ throw new IllegalStateException("Metadata node must be specified.");
+ }
+
+ Node metadataNode = ClusterProperties.INSTANCE.getNodeById(cluster.getMetadataNode());
+ if (metadataNode == null) {
+ throw new IllegalStateException("Invalid metadata node specified");
+ }
+
+ if (cluster.getHighAvailability().getFaultTolerance().getReplica() == null
+ || cluster.getHighAvailability().getFaultTolerance().getReplica().getNodeId() == null
+ || cluster.getHighAvailability().getFaultTolerance().getReplica().getNodeId().isEmpty()) {
+ throw new IllegalStateException("One or more replicas must be specified for metadata node.");
+ }
+
+ final Set<Replica> replicas = new HashSet<>();
+ for (String nodeId : cluster.getHighAvailability().getFaultTolerance().getReplica().getNodeId()) {
+ Node node = ClusterProperties.INSTANCE.getNodeById(nodeId);
+ if (node == null) {
+ throw new IllegalStateException("Invalid replica specified: " + nodeId);
+ }
+ replicas.add(new Replica(node));
+ }
+ MetadataOnlyReplicationStrategy st = new MetadataOnlyReplicationStrategy();
+ st.metadataNodeId = cluster.getMetadataNode();
+ st.metadataPrimaryReplica = new Replica(metadataNode);
+ st.metadataNodeReplicas = replicas;
+ return st;
+ }
+}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/ef173f34/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/replication/NoReplicationStrategy.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/replication/NoReplicationStrategy.java b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/replication/NoReplicationStrategy.java
new file mode 100644
index 0000000..43347f6
--- /dev/null
+++ b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/replication/NoReplicationStrategy.java
@@ -0,0 +1,52 @@
+/*
+ * 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 java.util.Collections;
+import java.util.Set;
+
+import org.apache.asterix.event.schema.cluster.Cluster;
+
+public class NoReplicationStrategy implements IReplicationStrategy {
+
+ @Override
+ public boolean isMatch(int datasetId) {
+ return false;
+ }
+
+ @Override
+ public boolean isParticipant(String nodeId) {
+ return false;
+ }
+
+ @Override
+ public Set<Replica> getRemotePrimaryReplicas(String nodeId) {
+ return Collections.emptySet();
+ }
+
+ @Override
+ public Set<Replica> getRemoteReplicas(String node) {
+ return Collections.emptySet();
+ }
+
+ @Override
+ public NoReplicationStrategy from(Cluster cluster) {
+ return new NoReplicationStrategy();
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/ef173f34/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/replication/Replica.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/replication/Replica.java b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/replication/Replica.java
index bd77778..267a22d 100644
--- a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/replication/Replica.java
+++ b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/replication/Replica.java
@@ -36,11 +36,13 @@ public class Replica {
UNKNOWN
}
- final Node node;
+ private final Node node;
private ReplicaState state = ReplicaState.UNKNOWN;
public Replica(Node node) {
- this.node = node;
+ this.node = new Node();
+ this.node.setId(node.getId());
+ this.node.setClusterIp(node.getClusterIp());
}
public ReplicaState getState() {
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/ef173f34/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/replication/ReplicationStrategyFactory.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/replication/ReplicationStrategyFactory.java b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/replication/ReplicationStrategyFactory.java
new file mode 100644
index 0000000..b61b38a
--- /dev/null
+++ b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/replication/ReplicationStrategyFactory.java
@@ -0,0 +1,62 @@
+/*
+ * 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 java.util.HashMap;
+import java.util.Map;
+
+import org.apache.asterix.event.schema.cluster.Cluster;
+
+public class ReplicationStrategyFactory {
+
+ private static final Map<String, Class<? extends IReplicationStrategy>>
+ BUILT_IN_REPLICATION_STRATEGY = new HashMap<>();
+
+ static {
+ BUILT_IN_REPLICATION_STRATEGY.put("no_replication", NoReplicationStrategy.class);
+ BUILT_IN_REPLICATION_STRATEGY.put("chained_declustering", ChainedDeclusteringReplicationStrategy.class);
+ BUILT_IN_REPLICATION_STRATEGY.put("metadata_only", MetadataOnlyReplicationStrategy.class);
+ }
+
+ private ReplicationStrategyFactory() {
+ throw new AssertionError();
+ }
+
+ public static IReplicationStrategy create(Cluster cluster) {
+ boolean highAvailabilityEnabled = cluster.getHighAvailability() != null
+ && cluster.getHighAvailability().getEnabled() != null
+ && Boolean.valueOf(cluster.getHighAvailability().getEnabled());
+
+ if (!highAvailabilityEnabled || cluster.getHighAvailability().getDataReplication() == null
+ || cluster.getHighAvailability().getDataReplication().getStrategy() == null) {
+ return new NoReplicationStrategy();
+ }
+ String strategyName = cluster.getHighAvailability().getDataReplication().getStrategy().toLowerCase();
+ if (!BUILT_IN_REPLICATION_STRATEGY.containsKey(strategyName)) {
+ throw new IllegalArgumentException(String.format("Unsupported Replication Strategy. Available types: %s",
+ BUILT_IN_REPLICATION_STRATEGY.keySet().toString()));
+ }
+ Class<? extends IReplicationStrategy> clazz = BUILT_IN_REPLICATION_STRATEGY.get(strategyName);
+ try {
+ return clazz.newInstance().from(cluster);
+ } catch (InstantiationException | IllegalAccessException e) {
+ throw new IllegalStateException(e);
+ }
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/ef173f34/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/storage/IndexFileProperties.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/storage/IndexFileProperties.java b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/storage/IndexFileProperties.java
new file mode 100644
index 0000000..ca6968f
--- /dev/null
+++ b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/storage/IndexFileProperties.java
@@ -0,0 +1,74 @@
+/*
+ * 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.storage;
+
+import java.io.File;
+
+import org.apache.asterix.common.utils.StoragePathUtil;
+
+/**
+ * A holder class for an index file properties.
+ */
+public class IndexFileProperties {
+
+ private final String fileName;
+ private final String idxName;
+ private final String dataverseName;
+ private final int partitionId;
+ private final int datasetId;
+
+ public IndexFileProperties(int partitionId, String dataverseName, String idxName, String fileName, int datasetId) {
+ this.partitionId = partitionId;
+ this.dataverseName = dataverseName;
+ this.idxName = idxName;
+ this.fileName = fileName;
+ this.datasetId = datasetId;
+ }
+
+ public String getFileName() {
+ return fileName;
+ }
+
+ public String getIdxName() {
+ return idxName;
+ }
+
+ public String getDataverseName() {
+ return dataverseName;
+ }
+
+ public int getPartitionId() {
+ return partitionId;
+ }
+
+ public int getDatasetId() {
+ return datasetId;
+ }
+
+ @Override
+ public String toString() {
+ StringBuilder sb = new StringBuilder();
+ sb.append(StoragePathUtil.PARTITION_DIR_PREFIX + partitionId + File.separator);
+ sb.append(dataverseName + File.separator);
+ sb.append(idxName + File.separator);
+ sb.append(fileName);
+ sb.append(" [Dataset ID: " + datasetId + "]");
+ return sb.toString();
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/ef173f34/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/transactions/IRecoveryManager.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/transactions/IRecoveryManager.java b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/transactions/IRecoveryManager.java
index 6816116..3e85276 100644
--- a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/transactions/IRecoveryManager.java
+++ b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/transactions/IRecoveryManager.java
@@ -34,6 +34,7 @@ import org.apache.hyracks.api.exceptions.HyracksDataException;
public interface IRecoveryManager {
public enum SystemState {
+ INITIAL_RUN,
NEW_UNIVERSE,
RECOVERING,
HEALTHY,
@@ -120,4 +121,6 @@ public interface IRecoveryManager {
* Deletes all temporary recovery files
*/
public void deleteRecoveryTemporaryFiles();
+
+ void startLocalRecovery(Set<Integer> partitions) throws IOException, ACIDException;
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/ef173f34/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/utils/StoragePathUtil.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/utils/StoragePathUtil.java b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/utils/StoragePathUtil.java
index 9f2e3e7..46cd476 100644
--- a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/utils/StoragePathUtil.java
+++ b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/utils/StoragePathUtil.java
@@ -78,6 +78,17 @@ public class StoragePathUtil {
}
/**
+ * @param fileAbsolutePath
+ * @return the file relative path starting from the partition directory
+ */
+ public static String getIndexFileRelativePath(String fileAbsolutePath) {
+ String[] tokens = fileAbsolutePath.split(File.separator);
+ //partition/dataverse/idx/fileName
+ return tokens[tokens.length - 4] + File.separator + tokens[tokens.length - 3] + File.separator
+ + tokens[tokens.length - 2] + File.separator + tokens[tokens.length - 1];
+ }
+
+ /**
* Create a file
* Note: this method is not thread safe. It is the responsibility of the caller to ensure no path conflict when
* creating files simultaneously
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/ef173f34/asterixdb/asterix-common/src/main/resources/schema/cluster.xsd
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-common/src/main/resources/schema/cluster.xsd b/asterixdb/asterix-common/src/main/resources/schema/cluster.xsd
index 79c377a..098b4e7 100644
--- a/asterixdb/asterix-common/src/main/resources/schema/cluster.xsd
+++ b/asterixdb/asterix-common/src/main/resources/schema/cluster.xsd
@@ -44,7 +44,7 @@
<xs:element name="http_port" type="xs:integer" />
<xs:element name="debug_port" type="xs:integer" />
<xs:element name="metadata_node" type="xs:string" />
- <xs:element name="enabled" type="xs:boolean" />
+ <xs:element name="enabled" type="xs:string" />
<xs:element name="replication_port" type="xs:integer" />
<xs:element name="replication_factor" type="xs:integer" />
<xs:element name="auto_failover" type="xs:boolean" />
@@ -57,6 +57,8 @@
<xs:element name="result_time_to_live" type="xs:long" />
<xs:element name="result_sweep_threshold" type="xs:long" />
<xs:element name="cc_root" type="xs:string" />
+ <xs:element name="strategy" type="xs:string" />
+ <xs:element name="node_id" type="xs:string" />
<!-- definition of complex elements -->
<xs:element name="working_dir">
@@ -87,15 +89,33 @@
<xs:element name="data_replication">
<xs:complexType>
<xs:sequence>
- <xs:element ref="cl:enabled" />
+ <xs:element ref="cl:strategy" />
<xs:element ref="cl:replication_port" />
<xs:element ref="cl:replication_factor" />
- <xs:element ref="cl:auto_failover" />
<xs:element ref="cl:replication_time_out" />
</xs:sequence>
</xs:complexType>
</xs:element>
+ <xs:element name="fault_tolerance">
+ <xs:complexType>
+ <xs:sequence>
+ <xs:element ref="cl:strategy" />
+ <xs:element ref="cl:replica" minOccurs="0"/>
+ </xs:sequence>
+ </xs:complexType>
+ </xs:element>
+
+ <xs:element name="high_availability">
+ <xs:complexType>
+ <xs:sequence>
+ <xs:element ref="cl:enabled" minOccurs="0"/>
+ <xs:element ref="cl:data_replication" minOccurs="0"/>
+ <xs:element ref="cl:fault_tolerance" minOccurs="0"/>
+ </xs:sequence>
+ </xs:complexType>
+ </xs:element>
+
<xs:element name="property">
<xs:complexType>
<xs:sequence>
@@ -136,6 +156,14 @@
</xs:complexType>
</xs:element>
+ <xs:element name="replica">
+ <xs:complexType>
+ <xs:sequence>
+ <xs:element ref="cl:node_id" maxOccurs="unbounded" />
+ </xs:sequence>
+ </xs:complexType>
+ </xs:element>
+
<xs:element name="cluster">
<xs:complexType>
<xs:sequence>
@@ -150,7 +178,7 @@
<xs:element ref="cl:iodevices" minOccurs="0" />
<xs:element ref="cl:working_dir" />
<xs:element ref="cl:metadata_node" />
- <xs:element ref="cl:data_replication" minOccurs="0" />
+ <xs:element ref="cl:high_availability" minOccurs="0" />
<xs:element ref="cl:master_node" />
<xs:element ref="cl:node" maxOccurs="unbounded" />
<xs:element ref="cl:substitute_nodes" />
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/ef173f34/asterixdb/asterix-common/src/test/java/org/apache/asterix/test/aql/TestExecutor.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-common/src/test/java/org/apache/asterix/test/aql/TestExecutor.java b/asterixdb/asterix-common/src/test/java/org/apache/asterix/test/aql/TestExecutor.java
index 288a739..8213213 100644
--- a/asterixdb/asterix-common/src/test/java/org/apache/asterix/test/aql/TestExecutor.java
+++ b/asterixdb/asterix-common/src/test/java/org/apache/asterix/test/aql/TestExecutor.java
@@ -76,7 +76,6 @@ import org.apache.http.impl.client.StandardHttpRequestRetryHandler;
import org.apache.http.util.EntityUtils;
import org.apache.hyracks.util.StorageUtil;
import org.junit.Assert;
-
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.DeserializationFeature;
import com.fasterxml.jackson.databind.JsonMappingException;
@@ -1108,6 +1107,21 @@ public class TestExecutor {
}
ProcessBuilder pb = new ProcessBuilder("kill", "-9", Integer.toString(nodePid));
pb.start().waitFor();
+ // Delete NC's transaction logs to re-initialize it as a new NC.
+ deleteNCTxnLogs(nodeId, cUnit);
+ }
+
+ private void deleteNCTxnLogs(String nodeId, CompilationUnit cUnit) throws Exception {
+ OutputFormat fmt = OutputFormat.forCompilationUnit(cUnit);
+ String endpoint = "/admin/cluster";
+ InputStream executeJSONGet = executeJSONGet(fmt, new URI("http://" + host + ":" + port + endpoint));
+ StringWriter actual = new StringWriter();
+ IOUtils.copy(executeJSONGet, actual, StandardCharsets.UTF_8);
+ String config = actual.toString();
+ ObjectMapper om = new ObjectMapper();
+ String logDir = om.readTree(config).findPath("transaction.log.dirs").get(nodeId).asText();
+ ProcessBuilder pb = new ProcessBuilder("rm", "-rf", logDir);
+ pb.start().waitFor();
}
public void executeTest(String actualPath, TestCaseContext testCaseCtx, ProcessBuilder pb,
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/ef173f34/asterixdb/asterix-installer/src/main/java/org/apache/asterix/installer/command/ConfigureCommand.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-installer/src/main/java/org/apache/asterix/installer/command/ConfigureCommand.java b/asterixdb/asterix-installer/src/main/java/org/apache/asterix/installer/command/ConfigureCommand.java
index 6d2b4b9..eac7586 100644
--- a/asterixdb/asterix-installer/src/main/java/org/apache/asterix/installer/command/ConfigureCommand.java
+++ b/asterixdb/asterix-installer/src/main/java/org/apache/asterix/installer/command/ConfigureCommand.java
@@ -44,7 +44,8 @@ public class ConfigureCommand extends AbstractCommand {
@Override
protected void execCommand() throws Exception {
configureCluster("local", "local.xml");
- configureCluster("local", "local_with_replication.xml");
+ configureCluster("local", "local_chained_declustering_rep.xml");
+ configureCluster("local", "local_metadata_only_rep.xml");
configureCluster("demo", "demo.xml");
String installerConfPath = InstallerDriver.getManagixHome() + File.separator + InstallerDriver.MANAGIX_CONF_XML;
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/ef173f34/asterixdb/asterix-installer/src/main/java/org/apache/asterix/installer/command/ValidateCommand.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-installer/src/main/java/org/apache/asterix/installer/command/ValidateCommand.java b/asterixdb/asterix-installer/src/main/java/org/apache/asterix/installer/command/ValidateCommand.java
index 4037eaf..748d811 100644
--- a/asterixdb/asterix-installer/src/main/java/org/apache/asterix/installer/command/ValidateCommand.java
+++ b/asterixdb/asterix-installer/src/main/java/org/apache/asterix/installer/command/ValidateCommand.java
@@ -262,40 +262,16 @@ public class ValidateCommand extends AbstractCommand {
boolean valid = true;
//if replication is disabled, no need to validate the settings
- if (cluster.getDataReplication() != null && cluster.getDataReplication().isEnabled()) {
-
- if (cluster.getDataReplication().getReplicationFactor() == null) {
- if (cluster.getNode().size() >= 3) {
- LOGGER.warn("Replication factor not defined. Using default value (3) " + WARNING);
-
- } else {
- valid = false;
- LOGGER.fatal("Replication factor not defined for data repliaction. " + ERROR);
- }
-
- }
-
- //replication factor = 1 means no replication
- if (cluster.getDataReplication().getReplicationFactor().intValue() == 1) {
- LOGGER.warn("Replication factor is set to 1. Disabling data replication" + WARNING);
- return true;
- }
-
- if (cluster.getDataReplication().getReplicationFactor().intValue() > cluster.getNode().size()) {
- LOGGER.fatal("Replication factor = " + cluster.getDataReplication().getReplicationFactor().intValue()
- + " requires at least " + cluster.getDataReplication().getReplicationFactor().intValue()
- + " nodes in the cluster" + ERROR);
- valid = false;
- }
-
- if (cluster.getDataReplication().getReplicationPort() == null
- || cluster.getDataReplication().getReplicationPort().toString().length() == 0) {
+ if (cluster.getHighAvailability() != null && cluster.getHighAvailability().getDataReplication() != null) {
+ if (cluster.getHighAvailability().getDataReplication().getReplicationPort() == null || cluster
+ .getHighAvailability().getDataReplication().getReplicationPort().toString().length() == 0) {
valid = false;
LOGGER.fatal("Replication data port not defined for data repliaction. " + ERROR);
}
- if (cluster.getDataReplication().getReplicationTimeOut() == null
- || (cluster.getDataReplication().getReplicationTimeOut().intValue() + "").length() == 0) {
+ if (cluster.getHighAvailability().getDataReplication().getReplicationTimeOut() == null || String
+ .valueOf(cluster.getHighAvailability().getDataReplication().getReplicationTimeOut().intValue())
+ .length() == 0) {
LOGGER.warn("Replication maximum wait time not defined. Using default value (60 seconds) " + WARNING);
}