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 ji...@apache.org on 2014/09/11 22:51:01 UTC
[10/11] git commit: HDFS-7034. Archival Storage: Fix
TestBlockPlacement and TestStorageMover. Contributed by Jing Zhao.
HDFS-7034. Archival Storage: Fix TestBlockPlacement and TestStorageMover. Contributed by Jing Zhao.
Project: http://git-wip-us.apache.org/repos/asf/hadoop/repo
Commit: http://git-wip-us.apache.org/repos/asf/hadoop/commit/0d85f7e5
Tree: http://git-wip-us.apache.org/repos/asf/hadoop/tree/0d85f7e5
Diff: http://git-wip-us.apache.org/repos/asf/hadoop/diff/0d85f7e5
Branch: refs/heads/HDFS-6584
Commit: 0d85f7e59146cc3e9a040c2203995f3efd8ed4eb
Parents: 70dfe9c
Author: Jing Zhao <ji...@apache.org>
Authored: Thu Sep 11 13:00:43 2014 -0700
Committer: Jing Zhao <ji...@apache.org>
Committed: Thu Sep 11 13:00:43 2014 -0700
----------------------------------------------------------------------
.../org/apache/hadoop/hdfs/DFSConfigKeys.java | 2 +-
.../hadoop/hdfs/server/balancer/Dispatcher.java | 27 +-
.../hdfs/server/balancer/NameNodeConnector.java | 20 +-
.../server/blockmanagement/BlockManager.java | 4 +-
.../hadoop/hdfs/server/datanode/DataNode.java | 4 +-
.../apache/hadoop/hdfs/server/mover/Mover.java | 10 +-
.../hadoop/hdfs/TestBlockStoragePolicy.java | 8 +-
.../hdfs/server/mover/TestStorageMover.java | 286 ++++++++++---------
8 files changed, 208 insertions(+), 153 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hadoop/blob/0d85f7e5/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/DFSConfigKeys.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/DFSConfigKeys.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/DFSConfigKeys.java
index f3edbf5..641fb52 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/DFSConfigKeys.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/DFSConfigKeys.java
@@ -372,7 +372,7 @@ public class DFSConfigKeys extends CommonConfigurationKeys {
public static final int DFS_BALANCER_MOVERTHREADS_DEFAULT = 1000;
public static final String DFS_BALANCER_DISPATCHERTHREADS_KEY = "dfs.balancer.dispatcherThreads";
public static final int DFS_BALANCER_DISPATCHERTHREADS_DEFAULT = 200;
-
+
public static final String DFS_MOVER_MOVEDWINWIDTH_KEY = "dfs.mover.movedWinWidth";
public static final long DFS_MOVER_MOVEDWINWIDTH_DEFAULT = 5400*1000L;
public static final String DFS_MOVER_MOVERTHREADS_KEY = "dfs.mover.moverThreads";
http://git-wip-us.apache.org/repos/asf/hadoop/blob/0d85f7e5/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/balancer/Dispatcher.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/balancer/Dispatcher.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/balancer/Dispatcher.java
index 98bd58e..f2a1299 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/balancer/Dispatcher.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/balancer/Dispatcher.java
@@ -49,6 +49,7 @@ import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.classification.InterfaceAudience;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.CommonConfigurationKeys;
+import org.apache.hadoop.hdfs.DFSConfigKeys;
import org.apache.hadoop.hdfs.DFSUtil;
import org.apache.hadoop.hdfs.DistributedFileSystem;
import org.apache.hadoop.hdfs.StorageType;
@@ -88,7 +89,11 @@ public class Dispatcher {
private static final long MAX_BLOCKS_SIZE_TO_FETCH = 2 * GB;
private static final int MAX_NO_PENDING_MOVE_ITERATIONS = 5;
- private static final long DELAY_AFTER_ERROR = 10 * 1000L; // 10 seconds
+ /**
+ * the period of time to delay the usage of a DataNode after hitting
+ * errors when using it for migrating data
+ */
+ private static long delayAfterErrors = 10 * 1000;
private final NameNodeConnector nnc;
private final SaslDataTransferClient saslClient;
@@ -112,6 +117,7 @@ public class Dispatcher {
private final ExecutorService moveExecutor;
private final ExecutorService dispatchExecutor;
+
/** The maximum number of concurrent blocks moves at a datanode */
private final int maxConcurrentMovesPerNode;
@@ -187,10 +193,12 @@ public class Dispatcher {
@Override
public String toString() {
- final Block b = block.getBlock();
- return b + " with size=" + b.getNumBytes() + " from "
- + source.getDisplayName() + " to " + target.getDisplayName()
- + " through " + proxySource.datanode;
+ final Block b = block != null ? block.getBlock() : null;
+ String bStr = b != null ? (b + " with size=" + b.getNumBytes() + " ")
+ : " ";
+ return bStr + "from " + source.getDisplayName() + " to " + target
+ .getDisplayName() + " through " + (proxySource != null ? proxySource
+ .datanode : "");
}
/**
@@ -316,8 +324,8 @@ public class Dispatcher {
// further in order to avoid a potential storm of "threads quota
// exceeded" warnings when the dispatcher gets out of sync with work
// going on in datanodes.
- proxySource.activateDelay(DELAY_AFTER_ERROR);
- target.getDDatanode().activateDelay(DELAY_AFTER_ERROR);
+ proxySource.activateDelay(delayAfterErrors);
+ target.getDDatanode().activateDelay(delayAfterErrors);
} finally {
IOUtils.closeStream(out);
IOUtils.closeStream(in);
@@ -1043,6 +1051,11 @@ public class Dispatcher {
blockMoveWaitTime = time;
}
+ @VisibleForTesting
+ public static void setDelayAfterErrors(long time) {
+ delayAfterErrors = time;
+ }
+
/** shutdown thread pools */
public void shutdownNow() {
if (dispatchExecutor != null) {
http://git-wip-us.apache.org/repos/asf/hadoop/blob/0d85f7e5/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/balancer/NameNodeConnector.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/balancer/NameNodeConnector.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/balancer/NameNodeConnector.java
index 79815c0..9e08d51 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/balancer/NameNodeConnector.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/balancer/NameNodeConnector.java
@@ -25,13 +25,16 @@ import java.net.InetAddress;
import java.net.URI;
import java.util.*;
+import com.google.common.annotations.VisibleForTesting;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.classification.InterfaceAudience;
import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.FsServerDefaults;
import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hdfs.DFSConfigKeys;
import org.apache.hadoop.hdfs.DistributedFileSystem;
import org.apache.hadoop.hdfs.NameNodeProxies;
import org.apache.hadoop.hdfs.protocol.AlreadyBeingCreatedException;
@@ -53,6 +56,7 @@ public class NameNodeConnector implements Closeable {
private static final Log LOG = LogFactory.getLog(NameNodeConnector.class);
private static final int MAX_NOT_CHANGED_ITERATIONS = 5;
+ private static boolean createIdFile = true;
/** Create {@link NameNodeConnector} for the given namenodes. */
public static List<NameNodeConnector> newNameNodeConnectors(
@@ -83,6 +87,11 @@ public class NameNodeConnector implements Closeable {
return connectors;
}
+ @VisibleForTesting
+ public static void setCreateIdFile(boolean create) {
+ createIdFile = create;
+ }
+
private final URI nameNodeUri;
private final String blockpoolID;
@@ -117,9 +126,10 @@ public class NameNodeConnector implements Closeable {
final FsServerDefaults defaults = fs.getServerDefaults(new Path("/"));
this.keyManager = new KeyManager(blockpoolID, namenode,
defaults.getEncryptDataTransfer(), conf);
- // Exit if there is another one running.
- out = checkAndMarkRunning();
- if (out == null) {
+ // if it is for test, we do not create the id file
+ out = createIdFile ? checkAndMarkRunning() : null;
+ if (createIdFile && out == null) {
+ // Exit if there is another one running.
throw new IOException("Another " + name + " is running.");
}
}
@@ -188,9 +198,9 @@ public class NameNodeConnector implements Closeable {
*/
private OutputStream checkAndMarkRunning() throws IOException {
try {
- final DataOutputStream out = fs.create(idPath);
+ final FSDataOutputStream out = fs.create(idPath);
out.writeBytes(InetAddress.getLocalHost().getHostName());
- out.flush();
+ out.hflush();
return out;
} catch(RemoteException e) {
if(AlreadyBeingCreatedException.class.getName().equals(e.getClassName())){
http://git-wip-us.apache.org/repos/asf/hadoop/blob/0d85f7e5/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/blockmanagement/BlockManager.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/blockmanagement/BlockManager.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/blockmanagement/BlockManager.java
index 956900d..cb303a7 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/blockmanagement/BlockManager.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/blockmanagement/BlockManager.java
@@ -1502,7 +1502,7 @@ public class BlockManager {
* @throws IOException
* if the number of targets < minimum replication.
* @see BlockPlacementPolicy#chooseTarget(String, int, Node,
- * List, boolean, Set, long, StorageType)
+ * Set, long, List, BlockStoragePolicy)
*/
public DatanodeStorageInfo[] chooseTarget4NewBlock(final String src,
final int numOfReplicas, final DatanodeDescriptor client,
@@ -2811,7 +2811,7 @@ public class BlockManager {
return false; // only consider delHint for the first case
} else if (delHint == null) {
return false; // no delHint
- } else if (!excessTypes.remove(delHint.getStorageType())) {
+ } else if (!excessTypes.contains(delHint.getStorageType())) {
return false; // delHint storage type is not an excess type
} else {
// check if removing delHint reduces the number of racks
http://git-wip-us.apache.org/repos/asf/hadoop/blob/0d85f7e5/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/DataNode.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/DataNode.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/DataNode.java
index 3810621..9b57262 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/DataNode.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/DataNode.java
@@ -1744,7 +1744,9 @@ public class DataNode extends Configured
+ b + " (numBytes=" + b.getNumBytes() + ")"
+ ", stage=" + stage
+ ", clientname=" + clientname
- + ", targets=" + Arrays.asList(targets));
+ + ", targets=" + Arrays.asList(targets)
+ + ", target storage types=" + (targetStorageTypes == null ? "[]" :
+ Arrays.asList(targetStorageTypes)));
}
this.targets = targets;
this.targetStorageTypes = targetStorageTypes;
http://git-wip-us.apache.org/repos/asf/hadoop/blob/0d85f7e5/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/mover/Mover.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/mover/Mover.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/mover/Mover.java
index c5d6dab..96588ff 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/mover/Mover.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/mover/Mover.java
@@ -328,8 +328,6 @@ public class Mover {
if (scheduleMoves4Block(diff, lb)) {
hasRemaining |= (diff.existing.size() > 1 &&
diff.expected.size() > 1);
- } else {
- hasRemaining = false; // not able to schedule any move
}
}
}
@@ -453,9 +451,11 @@ public class Mover {
static int run(Map<URI, List<Path>> namenodes, Configuration conf)
throws IOException, InterruptedException {
- final long sleeptime = 2000 * conf.getLong(
- DFSConfigKeys.DFS_HEARTBEAT_INTERVAL_KEY,
- DFSConfigKeys.DFS_HEARTBEAT_INTERVAL_DEFAULT);
+ final long sleeptime =
+ conf.getLong(DFSConfigKeys.DFS_HEARTBEAT_INTERVAL_KEY,
+ DFSConfigKeys.DFS_HEARTBEAT_INTERVAL_DEFAULT) * 2000 +
+ conf.getLong(DFSConfigKeys.DFS_NAMENODE_REPLICATION_INTERVAL_KEY,
+ DFSConfigKeys.DFS_NAMENODE_REPLICATION_INTERVAL_DEFAULT) * 1000;
LOG.info("namenodes = " + namenodes);
List<NameNodeConnector> connectors = Collections.emptyList();
http://git-wip-us.apache.org/repos/asf/hadoop/blob/0d85f7e5/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/TestBlockStoragePolicy.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/TestBlockStoragePolicy.java b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/TestBlockStoragePolicy.java
index da7306c..158c225 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/TestBlockStoragePolicy.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/TestBlockStoragePolicy.java
@@ -256,7 +256,7 @@ public class TestBlockStoragePolicy {
final short replication = 3;
{
- final List<StorageType> chosen = Arrays.asList();
+ final List<StorageType> chosen = Lists.newArrayList();
method.checkChooseStorageTypes(hot, replication, chosen,
StorageType.DISK, StorageType.DISK, StorageType.DISK);
method.checkChooseStorageTypes(warm, replication, chosen,
@@ -393,7 +393,7 @@ public class TestBlockStoragePolicy {
final EnumSet<StorageType> unavailables = disk;
final boolean isNewBlock = true;
{
- final List<StorageType> chosen = Arrays.asList();
+ final List<StorageType> chosen = Lists.newArrayList();
checkChooseStorageTypes(hot, replication, chosen, unavailables, isNewBlock);
checkChooseStorageTypes(warm, replication, chosen, unavailables, isNewBlock,
StorageType.ARCHIVE, StorageType.ARCHIVE, StorageType.ARCHIVE);
@@ -500,7 +500,7 @@ public class TestBlockStoragePolicy {
final short replication = 3;
{
- final List<StorageType> chosen = Arrays.asList();
+ final List<StorageType> chosen = Lists.newArrayList();
method.checkChooseStorageTypes(hot, replication, chosen,
StorageType.DISK, StorageType.DISK, StorageType.DISK);
method.checkChooseStorageTypes(warm, replication, chosen,
@@ -603,7 +603,7 @@ public class TestBlockStoragePolicy {
final EnumSet<StorageType> unavailables = disk;
final boolean isNewBlock = false;
{
- final List<StorageType> chosen = Arrays.asList();
+ final List<StorageType> chosen = Lists.newArrayList();
checkChooseStorageTypes(hot, replication, chosen, unavailables, isNewBlock,
StorageType.ARCHIVE, StorageType.ARCHIVE, StorageType.ARCHIVE);
checkChooseStorageTypes(warm, replication, chosen, unavailables, isNewBlock,
http://git-wip-us.apache.org/repos/asf/hadoop/blob/0d85f7e5/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/mover/TestStorageMover.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/mover/TestStorageMover.java b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/mover/TestStorageMover.java
index fda744f..d5d5cab 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/mover/TestStorageMover.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/mover/TestStorageMover.java
@@ -44,9 +44,13 @@ import org.apache.hadoop.hdfs.protocol.HdfsConstants;
import org.apache.hadoop.hdfs.protocol.HdfsFileStatus;
import org.apache.hadoop.hdfs.protocol.HdfsLocatedFileStatus;
import org.apache.hadoop.hdfs.protocol.LocatedBlock;
+import org.apache.hadoop.hdfs.protocol.datatransfer.DataTransferProtocol;
import org.apache.hadoop.hdfs.server.balancer.Dispatcher;
import org.apache.hadoop.hdfs.server.balancer.ExitStatus;
+import org.apache.hadoop.hdfs.server.balancer.NameNodeConnector;
import org.apache.hadoop.hdfs.server.blockmanagement.BlockPlacementPolicy;
+import org.apache.hadoop.hdfs.server.datanode.DataNode;
+import org.apache.hadoop.hdfs.server.datanode.DataNodeTestUtils;
import org.apache.hadoop.hdfs.server.namenode.snapshot.SnapshotTestHelper;
import org.apache.hadoop.io.IOUtils;
import org.apache.log4j.Level;
@@ -66,6 +70,8 @@ public class TestStorageMover {
).getLogger().setLevel(Level.ALL);
((Log4JLogger)LogFactory.getLog(Dispatcher.class)
).getLogger().setLevel(Level.ALL);
+ ((Log4JLogger)LogFactory.getLog(DataTransferProtocol.class)).getLogger()
+ .setLevel(Level.ALL);
}
private static final int BLOCK_SIZE = 1024;
@@ -80,6 +86,8 @@ public class TestStorageMover {
static {
DEFAULT_CONF.setLong(DFSConfigKeys.DFS_BLOCK_SIZE_KEY, BLOCK_SIZE);
DEFAULT_CONF.setLong(DFSConfigKeys.DFS_HEARTBEAT_INTERVAL_KEY, 1L);
+ DEFAULT_CONF.setLong(DFSConfigKeys.DFS_NAMENODE_REPLICATION_INTERVAL_KEY,
+ 2L);
DEFAULT_CONF.setLong(DFSConfigKeys.DFS_MOVER_MOVEDWINWIDTH_KEY, 2000L);
DEFAULT_POLICIES = BlockStoragePolicy.readBlockStorageSuite(DEFAULT_CONF);
@@ -87,6 +95,9 @@ public class TestStorageMover {
WARM = DEFAULT_POLICIES.getPolicy("WARM");
COLD = DEFAULT_POLICIES.getPolicy("COLD");
Dispatcher.setBlockMoveWaitTime(1000L);
+ Dispatcher.setDelayAfterErrors(1000L);
+ // do not create id file since we will eat up all the disk space
+ NameNodeConnector.setCreateIdFile(false);
}
/**
@@ -151,7 +162,7 @@ public class TestStorageMover {
ClusterScheme() {
this(DEFAULT_CONF, NUM_DATANODES, REPL,
- genStorageTypes(NUM_DATANODES, 1, 1), null);
+ genStorageTypes(NUM_DATANODES), null);
}
ClusterScheme(Configuration conf, int numDataNodes, short repl,
@@ -195,7 +206,7 @@ public class TestStorageMover {
dfs = cluster.getFileSystem();
}
- private void runBasicTest(boolean shotdown) throws Exception {
+ private void runBasicTest(boolean shutdown) throws Exception {
setupCluster();
try {
prepareNamespace();
@@ -205,7 +216,7 @@ public class TestStorageMover {
migrate();
verify(true);
} finally {
- if (shotdown) {
+ if (shutdown) {
shutdownCluster();
}
}
@@ -233,7 +244,7 @@ public class TestStorageMover {
/**
* Run the migration tool.
*/
- void migrate(String... args) throws Exception {
+ void migrate() throws Exception {
runMover();
Thread.sleep(5000); // let the NN finish deletion
}
@@ -242,6 +253,9 @@ public class TestStorageMover {
* Verify block locations after running the migration tool.
*/
void verify(boolean verifyAll) throws Exception {
+ for (DataNode dn : cluster.getDataNodes()) {
+ DataNodeTestUtils.triggerBlockReport(dn);
+ }
if (verifyAll) {
verifyNamespace();
} else {
@@ -308,7 +322,8 @@ public class TestStorageMover {
final Mover.StorageTypeDiff diff = new Mover.StorageTypeDiff(types,
lb.getStorageTypes());
Assert.assertTrue(fileStatus.getFullName(parent.toString())
- + " with policy " + policy + " has non-empty overlap: " + diff,
+ + " with policy " + policy + " has non-empty overlap: " + diff
+ + ", the corresponding block is " + lb.getBlock().getLocalBlock(),
diff.removeOverlap());
}
}
@@ -378,6 +393,7 @@ public class TestStorageMover {
return "[disk=" + disk + ", archive=" + archive + "]";
}
}
+
private static StorageType[][] genStorageTypes(int numDataNodes) {
return genStorageTypes(numDataNodes, 0, 0);
}
@@ -414,21 +430,6 @@ public class TestStorageMover {
return capacities;
}
- /**
- * A normal case for Mover: move a file into archival storage
- */
- @Test
- public void testMigrateFileToArchival() throws Exception {
- final Path foo = new Path("/foo");
- Map<Path, BlockStoragePolicy> policyMap = Maps.newHashMap();
- policyMap.put(foo, COLD);
- NamespaceScheme nsScheme = new NamespaceScheme(null, Arrays.asList(foo),
- 2*BLOCK_SIZE, null, policyMap);
- ClusterScheme clusterScheme = new ClusterScheme(DEFAULT_CONF,
- NUM_DATANODES, REPL, genStorageTypes(NUM_DATANODES), null);
- new MigrationTest(clusterScheme, nsScheme).runBasicTest(true);
- }
-
private static class PathPolicyMap {
final Map<Path, BlockStoragePolicy> map = Maps.newHashMap();
final Path hot = new Path("/hot");
@@ -447,13 +448,13 @@ public class TestStorageMover {
}
}
}
-
+
NamespaceScheme newNamespaceScheme() {
return new NamespaceScheme(Arrays.asList(hot, warm, cold),
files, BLOCK_SIZE/2, null, map);
}
-
- /**
+
+ /**
* Move hot files to warm and cold, warm files to hot and cold,
* and cold files to hot and warm.
*/
@@ -473,21 +474,41 @@ public class TestStorageMover {
}
/**
+ * A normal case for Mover: move a file into archival storage
+ */
+ @Test
+ public void testMigrateFileToArchival() throws Exception {
+ LOG.info("testMigrateFileToArchival");
+ final Path foo = new Path("/foo");
+ Map<Path, BlockStoragePolicy> policyMap = Maps.newHashMap();
+ policyMap.put(foo, COLD);
+ NamespaceScheme nsScheme = new NamespaceScheme(null, Arrays.asList(foo),
+ 2*BLOCK_SIZE, null, policyMap);
+ ClusterScheme clusterScheme = new ClusterScheme(DEFAULT_CONF,
+ NUM_DATANODES, REPL, genStorageTypes(NUM_DATANODES), null);
+ new MigrationTest(clusterScheme, nsScheme).runBasicTest(true);
+ }
+
+ /**
* Test directories with Hot, Warm and Cold polices.
*/
@Test
public void testHotWarmColdDirs() throws Exception {
+ LOG.info("testHotWarmColdDirs");
PathPolicyMap pathPolicyMap = new PathPolicyMap(3);
NamespaceScheme nsScheme = pathPolicyMap.newNamespaceScheme();
ClusterScheme clusterScheme = new ClusterScheme();
MigrationTest test = new MigrationTest(clusterScheme, nsScheme);
- test.runBasicTest(false);
+ try {
+ test.runBasicTest(false);
+ pathPolicyMap.moveAround(test.dfs);
+ test.migrate();
- pathPolicyMap.moveAround(test.dfs);
- test.migrate();
- test.verify(true);
- test.shutdownCluster();
+ test.verify(true);
+ } finally {
+ test.shutdownCluster();
+ }
}
/**
@@ -495,76 +516,81 @@ public class TestStorageMover {
*/
@Test
public void testNoSpaceDisk() throws Exception {
+ LOG.info("testNoSpaceDisk");
final PathPolicyMap pathPolicyMap = new PathPolicyMap(0);
final NamespaceScheme nsScheme = pathPolicyMap.newNamespaceScheme();
- final long diskCapacity = (10 + HdfsConstants.MIN_BLOCKS_FOR_WRITE)*BLOCK_SIZE;
- final long archiveCapacity = 100*BLOCK_SIZE;
+ final long diskCapacity = (6 + HdfsConstants.MIN_BLOCKS_FOR_WRITE)
+ * BLOCK_SIZE;
+ final long archiveCapacity = 100 * BLOCK_SIZE;
final long[][] capacities = genCapacities(NUM_DATANODES, 1, 1,
diskCapacity, archiveCapacity);
- final ClusterScheme clusterScheme = new ClusterScheme(DEFAULT_CONF,
+ Configuration conf = new Configuration(DEFAULT_CONF);
+ final ClusterScheme clusterScheme = new ClusterScheme(conf,
NUM_DATANODES, REPL, genStorageTypes(NUM_DATANODES, 1, 1), capacities);
final MigrationTest test = new MigrationTest(clusterScheme, nsScheme);
- test.runBasicTest(false);
-
- // create hot files with replication 3 until not more spaces.
- final short replication = 3;
- {
- int hotFileCount = 0;
- try {
- for(; ; hotFileCount++) {
- final Path p = new Path(pathPolicyMap.hot, "file" + hotFileCount);
- DFSTestUtil.createFile(test.dfs, p, BLOCK_SIZE, replication, 0L);
+ try {
+ test.runBasicTest(false);
+
+ // create hot files with replication 3 until not more spaces.
+ final short replication = 3;
+ {
+ int hotFileCount = 0;
+ try {
+ for (; ; hotFileCount++) {
+ final Path p = new Path(pathPolicyMap.hot, "file" + hotFileCount);
+ DFSTestUtil.createFile(test.dfs, p, BLOCK_SIZE, replication, 0L);
+ }
+ } catch (IOException e) {
+ LOG.info("Expected: hotFileCount=" + hotFileCount, e);
}
- } catch(IOException e) {
- LOG.info("Expected: hotFileCount=" + hotFileCount, e);
+ Assert.assertTrue(hotFileCount >= 1);
}
- Assert.assertTrue(hotFileCount >= 2);
- }
- // create hot files with replication 1 to use up all remaining spaces.
- {
- int hotFileCount_r1 = 0;
- try {
- for(; ; hotFileCount_r1++) {
- final Path p = new Path(pathPolicyMap.hot, "file_r1_" + hotFileCount_r1);
- DFSTestUtil.createFile(test.dfs, p, BLOCK_SIZE, (short)1, 0L);
+ // create hot files with replication 1 to use up all remaining spaces.
+ {
+ int hotFileCount_r1 = 0;
+ try {
+ for (; ; hotFileCount_r1++) {
+ final Path p = new Path(pathPolicyMap.hot, "file_r1_" + hotFileCount_r1);
+ DFSTestUtil.createFile(test.dfs, p, BLOCK_SIZE, (short) 1, 0L);
+ }
+ } catch (IOException e) {
+ LOG.info("Expected: hotFileCount_r1=" + hotFileCount_r1, e);
}
- } catch(IOException e) {
- LOG.info("Expected: hotFileCount_r1=" + hotFileCount_r1, e);
}
- }
- { // test increasing replication. Since DISK is full,
- // new replicas should be stored in ARCHIVE as a fallback storage.
- final Path file0 = new Path(pathPolicyMap.hot, "file0");
- final Replication r = test.getReplication(file0);
- final short newReplication = (short)5;
- test.dfs.setReplication(file0, newReplication);
- Thread.sleep(10000);
- test.verifyReplication(file0, r.disk, newReplication - r.disk);
- }
+ { // test increasing replication. Since DISK is full,
+ // new replicas should be stored in ARCHIVE as a fallback storage.
+ final Path file0 = new Path(pathPolicyMap.hot, "file0");
+ final Replication r = test.getReplication(file0);
+ final short newReplication = (short) 5;
+ test.dfs.setReplication(file0, newReplication);
+ Thread.sleep(10000);
+ test.verifyReplication(file0, r.disk, newReplication - r.disk);
+ }
- { // test creating a cold file and then increase replication
- final Path p = new Path(pathPolicyMap.cold, "foo");
- DFSTestUtil.createFile(test.dfs, p, BLOCK_SIZE, replication, 0L);
- test.verifyReplication(p, 0, replication);
+ { // test creating a cold file and then increase replication
+ final Path p = new Path(pathPolicyMap.cold, "foo");
+ DFSTestUtil.createFile(test.dfs, p, BLOCK_SIZE, replication, 0L);
+ test.verifyReplication(p, 0, replication);
- final short newReplication = 5;
- test.dfs.setReplication(p, newReplication);
- Thread.sleep(10000);
- test.verifyReplication(p, 0, newReplication);
- }
+ final short newReplication = 5;
+ test.dfs.setReplication(p, newReplication);
+ Thread.sleep(10000);
+ test.verifyReplication(p, 0, newReplication);
+ }
- { //test move a hot file to warm
- final Path file1 = new Path(pathPolicyMap.hot, "file1");
- test.dfs.rename(file1, pathPolicyMap.warm);
- test.migrate();
- test.verifyFile(new Path(pathPolicyMap.warm, "file1"), WARM.getId());;
+ { //test move a hot file to warm
+ final Path file1 = new Path(pathPolicyMap.hot, "file1");
+ test.dfs.rename(file1, pathPolicyMap.warm);
+ test.migrate();
+ test.verifyFile(new Path(pathPolicyMap.warm, "file1"), WARM.getId());
+ }
+ } finally {
+ test.shutdownCluster();
}
-
- test.shutdownCluster();
}
/**
@@ -572,73 +598,77 @@ public class TestStorageMover {
*/
@Test
public void testNoSpaceArchive() throws Exception {
+ LOG.info("testNoSpaceArchive");
final PathPolicyMap pathPolicyMap = new PathPolicyMap(0);
final NamespaceScheme nsScheme = pathPolicyMap.newNamespaceScheme();
- final long diskCapacity = 100*BLOCK_SIZE;
- final long archiveCapacity = (10 + HdfsConstants.MIN_BLOCKS_FOR_WRITE)*BLOCK_SIZE;
+ final long diskCapacity = 100 * BLOCK_SIZE;
+ final long archiveCapacity = (6 + HdfsConstants.MIN_BLOCKS_FOR_WRITE)
+ * BLOCK_SIZE;
final long[][] capacities = genCapacities(NUM_DATANODES, 1, 1,
diskCapacity, archiveCapacity);
final ClusterScheme clusterScheme = new ClusterScheme(DEFAULT_CONF,
NUM_DATANODES, REPL, genStorageTypes(NUM_DATANODES, 1, 1), capacities);
final MigrationTest test = new MigrationTest(clusterScheme, nsScheme);
- test.runBasicTest(false);
-
- // create cold files with replication 3 until not more spaces.
- final short replication = 3;
- {
- int coldFileCount = 0;
- try {
- for(; ; coldFileCount++) {
- final Path p = new Path(pathPolicyMap.cold, "file" + coldFileCount);
- DFSTestUtil.createFile(test.dfs, p, BLOCK_SIZE, replication, 0L);
+ try {
+ test.runBasicTest(false);
+
+ // create cold files with replication 3 until not more spaces.
+ final short replication = 3;
+ {
+ int coldFileCount = 0;
+ try {
+ for (; ; coldFileCount++) {
+ final Path p = new Path(pathPolicyMap.cold, "file" + coldFileCount);
+ DFSTestUtil.createFile(test.dfs, p, BLOCK_SIZE, replication, 0L);
+ }
+ } catch (IOException e) {
+ LOG.info("Expected: coldFileCount=" + coldFileCount, e);
}
- } catch(IOException e) {
- LOG.info("Expected: coldFileCount=" + coldFileCount, e);
+ Assert.assertTrue(coldFileCount >= 1);
}
- Assert.assertTrue(coldFileCount >= 2);
- }
- // create cold files with replication 1 to use up all remaining spaces.
- {
- int coldFileCount_r1 = 0;
- try {
- for(; ; coldFileCount_r1++) {
- final Path p = new Path(pathPolicyMap.cold, "file_r1_" + coldFileCount_r1);
- DFSTestUtil.createFile(test.dfs, p, BLOCK_SIZE, (short)1, 0L);
+ // create cold files with replication 1 to use up all remaining spaces.
+ {
+ int coldFileCount_r1 = 0;
+ try {
+ for (; ; coldFileCount_r1++) {
+ final Path p = new Path(pathPolicyMap.cold, "file_r1_" + coldFileCount_r1);
+ DFSTestUtil.createFile(test.dfs, p, BLOCK_SIZE, (short) 1, 0L);
+ }
+ } catch (IOException e) {
+ LOG.info("Expected: coldFileCount_r1=" + coldFileCount_r1, e);
}
- } catch(IOException e) {
- LOG.info("Expected: coldFileCount_r1=" + coldFileCount_r1, e);
}
- }
- { // test increasing replication but new replicas cannot be created
- // since no more ARCHIVE space.
- final Path file0 = new Path(pathPolicyMap.cold, "file0");
- final Replication r = test.getReplication(file0);
- LOG.info("XXX " + file0 + ": replication=" + r);
- Assert.assertEquals(0, r.disk);
+ { // test increasing replication but new replicas cannot be created
+ // since no more ARCHIVE space.
+ final Path file0 = new Path(pathPolicyMap.cold, "file0");
+ final Replication r = test.getReplication(file0);
+ LOG.info("XXX " + file0 + ": replication=" + r);
+ Assert.assertEquals(0, r.disk);
- final short newReplication = (short)5;
- test.dfs.setReplication(file0, newReplication);
- Thread.sleep(10000);
+ final short newReplication = (short) 5;
+ test.dfs.setReplication(file0, newReplication);
+ Thread.sleep(10000);
- test.verifyReplication(file0, 0, r.archive);
- }
+ test.verifyReplication(file0, 0, r.archive);
+ }
- { // test creating a hot file
- final Path p = new Path(pathPolicyMap.hot, "foo");
- DFSTestUtil.createFile(test.dfs, p, BLOCK_SIZE, (short)3, 0L);
- }
+ { // test creating a hot file
+ final Path p = new Path(pathPolicyMap.hot, "foo");
+ DFSTestUtil.createFile(test.dfs, p, BLOCK_SIZE, (short) 3, 0L);
+ }
- { //test move a cold file to warm
- final Path file1 = new Path(pathPolicyMap.hot, "file1");
- test.dfs.rename(file1, pathPolicyMap.warm);
- test.migrate();
- test.verify(true);
+ { //test move a cold file to warm
+ final Path file1 = new Path(pathPolicyMap.cold, "file1");
+ test.dfs.rename(file1, pathPolicyMap.warm);
+ test.migrate();
+ test.verify(true);
+ }
+ } finally {
+ test.shutdownCluster();
}
-
- test.shutdownCluster();
}
}