You are viewing a plain text version of this content. The canonical link for it is here.
Posted to common-commits@hadoop.apache.org by sz...@apache.org on 2012/09/14 14:32:19 UTC
svn commit: r1384742 - in /hadoop/common/branches/branch-1: ./ src/hdfs/
src/hdfs/org/apache/hadoop/hdfs/ src/hdfs/org/apache/hadoop/hdfs/protocol/
src/hdfs/org/apache/hadoop/hdfs/server/namenode/
src/test/org/apache/hadoop/hdfs/
Author: szetszwo
Date: Fri Sep 14 12:32:18 2012
New Revision: 1384742
URL: http://svn.apache.org/viewvc?rev=1384742&view=rev
Log:
HDFS-3703. Datanodes are marked stale if heartbeat is not received in configured timeout and are selected as the last location to read from. Contributed by Jing Zhao
Modified:
hadoop/common/branches/branch-1/CHANGES.txt
hadoop/common/branches/branch-1/src/hdfs/hdfs-default.xml
hadoop/common/branches/branch-1/src/hdfs/org/apache/hadoop/hdfs/DFSConfigKeys.java
hadoop/common/branches/branch-1/src/hdfs/org/apache/hadoop/hdfs/DFSUtil.java
hadoop/common/branches/branch-1/src/hdfs/org/apache/hadoop/hdfs/protocol/DatanodeInfo.java
hadoop/common/branches/branch-1/src/hdfs/org/apache/hadoop/hdfs/server/namenode/FSNamesystem.java
hadoop/common/branches/branch-1/src/test/org/apache/hadoop/hdfs/TestGetBlocks.java
Modified: hadoop/common/branches/branch-1/CHANGES.txt
URL: http://svn.apache.org/viewvc/hadoop/common/branches/branch-1/CHANGES.txt?rev=1384742&r1=1384741&r2=1384742&view=diff
==============================================================================
--- hadoop/common/branches/branch-1/CHANGES.txt (original)
+++ hadoop/common/branches/branch-1/CHANGES.txt Fri Sep 14 12:32:18 2012
@@ -268,6 +268,10 @@ Release 1.1.0 - unreleased
HADOOP-7823. Port HADOOP-4012 providing split support for bzip2 compressed
files to branch-1. (Andrew Purtell via cdouglas)
+ HDFS-3703. Datanodes are marked stale if heartbeat is not received in
+ configured timeout and are selected as the last location to read from.
+ (Jing Zhao via szetszwo)
+
IMPROVEMENTS
HADOOP-8656. Backport forced daemon shutdown of HADOOP-8353 into branch-1
Modified: hadoop/common/branches/branch-1/src/hdfs/hdfs-default.xml
URL: http://svn.apache.org/viewvc/hadoop/common/branches/branch-1/src/hdfs/hdfs-default.xml?rev=1384742&r1=1384741&r2=1384742&view=diff
==============================================================================
--- hadoop/common/branches/branch-1/src/hdfs/hdfs-default.xml (original)
+++ hadoop/common/branches/branch-1/src/hdfs/hdfs-default.xml Fri Sep 14 12:32:18 2012
@@ -612,4 +612,28 @@ creations/deletions), or "all".</descrip
</description>
</property>
+<property>
+ <name>dfs.namenode.check.stale.datanode</name>
+ <value>false</value>
+ <description>
+ Indicate whether or not to check "stale" datanodes whose
+ heartbeat messages have not been received by the namenode
+ for more than a specified time interval. If this configuration
+ parameter is set as true, the stale datanodes will be moved to
+ the end of the target node list for reading. The writing will
+ also try to avoid stale nodes.
+ </description>
+</property>
+
+<property>
+ <name>dfs.namenode.stale.datanode.interval</name>
+ <value>30000</value>
+ <description>
+ Default time interval for marking a datanode as "stale", i.e., if
+ the namenode has not received heartbeat msg from a datanode for
+ more than this time interval, the datanode will be marked and treated
+ as "stale" by default.
+ </description>
+</property>
+
</configuration>
Modified: hadoop/common/branches/branch-1/src/hdfs/org/apache/hadoop/hdfs/DFSConfigKeys.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/branch-1/src/hdfs/org/apache/hadoop/hdfs/DFSConfigKeys.java?rev=1384742&r1=1384741&r2=1384742&view=diff
==============================================================================
--- hadoop/common/branches/branch-1/src/hdfs/org/apache/hadoop/hdfs/DFSConfigKeys.java (original)
+++ hadoop/common/branches/branch-1/src/hdfs/org/apache/hadoop/hdfs/DFSConfigKeys.java Fri Sep 14 12:32:18 2012
@@ -146,6 +146,13 @@ public class DFSConfigKeys extends Commo
public static final String DFS_NAMENODE_CHECKPOINT_DIR_KEY = "dfs.namenode.checkpoint.dir";
public static final String DFS_NAMENODE_CHECKPOINT_EDITS_DIR_KEY = "dfs.namenode.checkpoint.edits.dir";
public static final String DFS_CLIENT_LOCAL_INTERFACES = "dfs.client.local.interfaces";
+
+ // Whether to enable datanode's stale state detection and usage
+ public static final String DFS_NAMENODE_CHECK_STALE_DATANODE_KEY = "dfs.namenode.check.stale.datanode";
+ public static final boolean DFS_NAMENODE_CHECK_STALE_DATANODE_DEFAULT = false;
+ // The default value of the time interval for marking datanodes as stale
+ public static final String DFS_NAMENODE_STALE_DATANODE_INTERVAL_KEY = "dfs.namenode.stale.datanode.interval";
+ public static final long DFS_NAMENODE_STALE_DATANODE_INTERVAL_MILLI_DEFAULT = 30 * 1000; // 30s
//Code in hdfs is not updated to use these keys.
public static final String DFS_CLIENT_BLOCK_WRITE_LOCATEFOLLOWINGBLOCK_RETRIES_KEY = "dfs.client.block.write.locateFollowingBlock.retries";
Modified: hadoop/common/branches/branch-1/src/hdfs/org/apache/hadoop/hdfs/DFSUtil.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/branch-1/src/hdfs/org/apache/hadoop/hdfs/DFSUtil.java?rev=1384742&r1=1384741&r2=1384742&view=diff
==============================================================================
--- hadoop/common/branches/branch-1/src/hdfs/org/apache/hadoop/hdfs/DFSUtil.java (original)
+++ hadoop/common/branches/branch-1/src/hdfs/org/apache/hadoop/hdfs/DFSUtil.java Fri Sep 14 12:32:18 2012
@@ -22,6 +22,7 @@ import java.io.UnsupportedEncodingExcept
import java.net.InetSocketAddress;
import java.net.URI;
import java.net.URISyntaxException;
+import java.util.Comparator;
import java.util.StringTokenizer;
import org.apache.hadoop.conf.Configuration;
@@ -167,5 +168,32 @@ public class DFSUtil {
}
return blocksReplWorkMultiplier;
}
+
+ /**
+ * Comparator for sorting DataNodeInfo[] based on stale states. Stale nodes
+ * are moved to the end of the array on sorting with this comparator.
+ */
+ public static class StaleComparator implements Comparator<DatanodeInfo> {
+ private long staleInterval;
+
+ /**
+ * Constructor of StaleComparator
+ *
+ * @param interval
+ * The time invertal for marking datanodes as stale is passed from
+ * outside, since the interval may be changed dynamically
+ */
+ public StaleComparator(long interval) {
+ this.staleInterval = interval;
+ }
+
+ @Override
+ public int compare(DatanodeInfo a, DatanodeInfo b) {
+ // Stale nodes will be moved behind the normal nodes
+ boolean aStale = a.isStale(staleInterval);
+ boolean bStale = b.isStale(staleInterval);
+ return aStale == bStale ? 0 : (aStale ? 1 : -1);
+ }
+ }
}
Modified: hadoop/common/branches/branch-1/src/hdfs/org/apache/hadoop/hdfs/protocol/DatanodeInfo.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/branch-1/src/hdfs/org/apache/hadoop/hdfs/protocol/DatanodeInfo.java?rev=1384742&r1=1384741&r2=1384742&view=diff
==============================================================================
--- hadoop/common/branches/branch-1/src/hdfs/org/apache/hadoop/hdfs/protocol/DatanodeInfo.java (original)
+++ hadoop/common/branches/branch-1/src/hdfs/org/apache/hadoop/hdfs/protocol/DatanodeInfo.java Fri Sep 14 12:32:18 2012
@@ -296,6 +296,23 @@ public class DatanodeInfo extends Datano
}
/**
+ * Check if the datanode is in stale state. Here if the namenode has not
+ * received heartbeat msg from a datanode for more than staleInterval (default
+ * value is
+ * {@link DFSConfigKeys#DFS_NAMENODE_STALE_DATANODE_INTERVAL_MILLI_DEFAULT}),
+ * the datanode will be treated as stale node.
+ *
+ * @param staleInterval
+ * the time interval for marking the node as stale. If the last
+ * update time is beyond the given time interval, the node will be
+ * marked as stale.
+ * @return true if the node is stale
+ */
+ public boolean isStale(long staleInterval) {
+ return (System.currentTimeMillis() - lastUpdate) >= staleInterval;
+ }
+
+ /**
* Retrieves the admin state of this node.
*/
public AdminStates getAdminState() {
Modified: hadoop/common/branches/branch-1/src/hdfs/org/apache/hadoop/hdfs/server/namenode/FSNamesystem.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/branch-1/src/hdfs/org/apache/hadoop/hdfs/server/namenode/FSNamesystem.java?rev=1384742&r1=1384741&r2=1384742&view=diff
==============================================================================
--- hadoop/common/branches/branch-1/src/hdfs/org/apache/hadoop/hdfs/server/namenode/FSNamesystem.java (original)
+++ hadoop/common/branches/branch-1/src/hdfs/org/apache/hadoop/hdfs/server/namenode/FSNamesystem.java Fri Sep 14 12:32:18 2012
@@ -365,6 +365,11 @@ public class FSNamesystem implements FSC
private long accessTimePrecision = 0;
private String nameNodeHostName;
+ /** Whether or not to check the stale datanodes */
+ private volatile boolean checkForStaleNodes;
+ /** The time interval for detecting stale datanodes */
+ private volatile long staleInterval;
+
/**
* FSNamesystem constructor.
*/
@@ -570,6 +575,22 @@ public class FSNamesystem implements FSC
+ " accessKeyUpdateInterval=" + accessKeyUpdateInterval / (60 * 1000)
+ " min(s), accessTokenLifetime=" + accessTokenLifetime / (60 * 1000)
+ " min(s)");
+
+ // set the value of stale interval based on configuration
+ this.checkForStaleNodes = conf.getBoolean(
+ DFSConfigKeys.DFS_NAMENODE_CHECK_STALE_DATANODE_KEY,
+ DFSConfigKeys.DFS_NAMENODE_CHECK_STALE_DATANODE_DEFAULT);
+ if (this.checkForStaleNodes) {
+ this.staleInterval = conf.getLong(
+ DFSConfigKeys.DFS_NAMENODE_STALE_DATANODE_INTERVAL_KEY,
+ DFSConfigKeys.DFS_NAMENODE_STALE_DATANODE_INTERVAL_MILLI_DEFAULT);
+ if (this.staleInterval < DFSConfigKeys.DFS_NAMENODE_STALE_DATANODE_INTERVAL_MILLI_DEFAULT) {
+ LOG.warn("The given interval for marking stale datanode = "
+ + this.staleInterval + ", which is smaller than the default value "
+ + DFSConfigKeys.DFS_NAMENODE_STALE_DATANODE_INTERVAL_MILLI_DEFAULT
+ + ".");
+ }
+ }
}
/**
@@ -933,8 +954,16 @@ public class FSNamesystem implements FSC
//sort the blocks
DatanodeDescriptor client = host2DataNodeMap.getDatanodeByHost(
clientMachine);
+ DFSUtil.StaleComparator comparator = null;
+ if (checkForStaleNodes) {
+ comparator = new DFSUtil.StaleComparator(staleInterval);
+ }
+ // Note: the last block is also included and sorted
for (LocatedBlock b : blocks.getLocatedBlocks()) {
clusterMap.pseudoSortByDistance(client, b.getLocations());
+ if (checkForStaleNodes) {
+ Arrays.sort(b.getLocations(), comparator);
+ }
}
}
return blocks;
@@ -4272,7 +4301,7 @@ public class FSNamesystem implements FSC
return getDatanodeListForReport(type).size();
}
- private synchronized ArrayList<DatanodeDescriptor> getDatanodeListForReport(
+ public synchronized ArrayList<DatanodeDescriptor> getDatanodeListForReport(
DatanodeReportType type) {
boolean listLiveNodes = type == DatanodeReportType.ALL ||
Modified: hadoop/common/branches/branch-1/src/test/org/apache/hadoop/hdfs/TestGetBlocks.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/branch-1/src/test/org/apache/hadoop/hdfs/TestGetBlocks.java?rev=1384742&r1=1384741&r2=1384742&view=diff
==============================================================================
--- hadoop/common/branches/branch-1/src/test/org/apache/hadoop/hdfs/TestGetBlocks.java (original)
+++ hadoop/common/branches/branch-1/src/test/org/apache/hadoop/hdfs/TestGetBlocks.java Fri Sep 14 12:32:18 2012
@@ -19,29 +19,115 @@ package org.apache.hadoop.hdfs;
import java.io.IOException;
import java.net.InetSocketAddress;
-import java.util.*;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Random;
+
+import junit.framework.TestCase;
import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FSDataOutputStream;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hdfs.protocol.Block;
+import org.apache.hadoop.hdfs.protocol.ClientProtocol;
import org.apache.hadoop.hdfs.protocol.DatanodeInfo;
+import org.apache.hadoop.hdfs.protocol.FSConstants.DatanodeReportType;
import org.apache.hadoop.hdfs.protocol.LocatedBlock;
+import org.apache.hadoop.hdfs.protocol.LocatedBlocks;
import org.apache.hadoop.hdfs.server.common.GenerationStamp;
+import org.apache.hadoop.hdfs.server.datanode.DataNode;
+import org.apache.hadoop.hdfs.server.namenode.DatanodeDescriptor;
import org.apache.hadoop.hdfs.server.namenode.NameNode;
-import org.apache.hadoop.hdfs.server.protocol.NamenodeProtocol;
import org.apache.hadoop.hdfs.server.protocol.BlocksWithLocations.BlockWithLocations;
-import org.apache.hadoop.fs.FSDataOutputStream;
-import org.apache.hadoop.fs.FileSystem;
-import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hdfs.server.protocol.NamenodeProtocol;
import org.apache.hadoop.ipc.RPC;
import org.apache.hadoop.ipc.RemoteException;
import org.apache.hadoop.net.NetUtils;
import org.apache.hadoop.security.UserGroupInformation;
-
-import junit.framework.TestCase;
+import org.junit.Test;
/**
* This class tests if block replacement request to data nodes work correctly.
*/
public class TestGetBlocks extends TestCase {
+
+ private static final int blockSize = 8192;
+ private static final String racks[] = new String[] { "/d1/r1", "/d1/r1",
+ "/d1/r2", "/d1/r2", "/d1/r2", "/d2/r3", "/d2/r3" };
+ private static final int numDatanodes = racks.length;
+
+ /**
+ * Test if the datanodes returned by
+ * {@link ClientProtocol#getBlockLocations(String, long, long)} is correct
+ * when 1) stale nodes checking is enabled, 2) a writing is going on,
+ * and 3) a datanode becomes stale happen simultaneously
+ *
+ * @throws Exception
+ */
+ public void testReadSelectNonStaleDatanode() throws Exception {
+ Configuration conf = new Configuration();
+ conf.setBoolean(DFSConfigKeys.DFS_NAMENODE_CHECK_STALE_DATANODE_KEY, true);
+ long staleInterval = 30 * 1000 * 60;
+ // DataNode will send out heartbeat every 15 minutes
+ // In this way, when we have set a datanode as stale,
+ // its heartbeat will not come to refresh its state
+ long heartbeatInterval = 15 * 60;
+ conf.setLong(DFSConfigKeys.DFS_NAMENODE_STALE_DATANODE_INTERVAL_KEY,
+ staleInterval);
+ conf.setLong("dfs.heartbeat.interval", heartbeatInterval);
+ MiniDFSCluster cluster = new MiniDFSCluster(conf, numDatanodes, true,
+ racks);
+
+ cluster.waitActive();
+ InetSocketAddress addr = new InetSocketAddress("localhost",
+ cluster.getNameNodePort());
+ DFSClient client = new DFSClient(addr, conf);
+ List<DatanodeDescriptor> nodeInfoList = cluster.getNameNode()
+ .getNamesystem().getDatanodeListForReport(DatanodeReportType.LIVE);
+ assertEquals("Unexpected number of datanodes", numDatanodes,
+ nodeInfoList.size());
+ FileSystem fileSys = cluster.getFileSystem();
+ FSDataOutputStream stm = null;
+ try {
+ // do the writing but do not close the FSDataOutputStream
+ // in order to mimic the ongoing writing
+ final Path fileName = new Path("/file1");
+ stm = fileSys.create(fileName, true,
+ fileSys.getConf().getInt("io.file.buffer.size", 4096), (short) 3,
+ blockSize);
+ stm.write(new byte[(blockSize * 3) / 2]);
+ // We do not close the stream so that
+ // the writing seems to be still ongoing
+ stm.sync();
+
+ LocatedBlocks blocks = cluster.getNameNode().getBlockLocations(
+ fileName.toString(), 0, blockSize);
+ DatanodeInfo[] nodes = blocks.get(0).getLocations();
+ assertEquals(nodes.length, 3);
+ for (DataNode dn : cluster.getDataNodes()) {
+ if (dn.getHostName().equals(nodes[0].getHostName())) {
+ // set the first node as stale
+ DatanodeDescriptor staleNodeInfo = cluster.getNameNode()
+ .getNamesystem().getDatanode(dn.dnRegistration);
+ staleNodeInfo.setLastUpdate(System.currentTimeMillis()
+ - staleInterval - 1);
+ }
+ }
+
+ LocatedBlocks blocksAfterStale = cluster.getNameNode().getBlockLocations(
+ fileName.toString(), 0, blockSize);
+ DatanodeInfo[] nodesAfterStale = blocksAfterStale.get(0).getLocations();
+ assertEquals(nodesAfterStale.length, 3);
+ assertEquals(nodesAfterStale[2].getHostName(), nodes[0].getHostName());
+ } finally {
+ if (stm != null) {
+ stm.close();
+ }
+ cluster.shutdown();
+ }
+ }
+
/** test getBlocks */
public void testGetBlocks() throws Exception {
final Configuration CONF = new Configuration();