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/08/27 19:39:45 UTC
git commit: HDFS-6911. Archival Storage: check if a block is already
scheduled in Mover. Contributed by Tsz Wo Nicholas Sze.
Repository: hadoop
Updated Branches:
refs/heads/HDFS-6584 555900a9d -> 8ea20b53a
HDFS-6911. Archival Storage: check if a block is already scheduled in Mover. Contributed by Tsz Wo Nicholas Sze.
Project: http://git-wip-us.apache.org/repos/asf/hadoop/repo
Commit: http://git-wip-us.apache.org/repos/asf/hadoop/commit/8ea20b53
Tree: http://git-wip-us.apache.org/repos/asf/hadoop/tree/8ea20b53
Diff: http://git-wip-us.apache.org/repos/asf/hadoop/diff/8ea20b53
Branch: refs/heads/HDFS-6584
Commit: 8ea20b53a861a2771c206afaacf8e7783568c4b1
Parents: 555900a
Author: Jing Zhao <ji...@apache.org>
Authored: Wed Aug 27 10:38:10 2014 -0700
Committer: Jing Zhao <ji...@apache.org>
Committed: Wed Aug 27 10:38:10 2014 -0700
----------------------------------------------------------------------
.../hadoop/hdfs/server/balancer/Dispatcher.java | 50 +++++++-----
.../hadoop/hdfs/server/balancer/Matcher.java | 15 ++++
.../apache/hadoop/hdfs/server/mover/Mover.java | 65 ++++++++-------
.../hadoop/hdfs/TestBlockStoragePolicy.java | 79 +++++++++++++-----
.../hadoop/hdfs/server/mover/TestMover.java | 85 ++++++++++++++++++++
5 files changed, 228 insertions(+), 66 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hadoop/blob/8ea20b53/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 bb498cc..41ea1f3 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
@@ -175,11 +175,7 @@ public class Dispatcher {
private DDatanode proxySource;
private StorageGroup target;
- private PendingMove() {
- }
-
- public PendingMove(DBlock block, Source source, StorageGroup target) {
- this.block = block;
+ private PendingMove(Source source, StorageGroup target) {
this.source = source;
this.target = target;
}
@@ -199,9 +195,11 @@ public class Dispatcher {
* @return true if a block and its proxy are chosen; false otherwise
*/
private boolean chooseBlockAndProxy() {
+ // source and target must have the same storage type
+ final StorageType t = source.getStorageType();
// iterate all source's blocks until find a good one
for (Iterator<DBlock> i = source.getBlockIterator(); i.hasNext();) {
- if (markMovedIfGoodBlock(i.next())) {
+ if (markMovedIfGoodBlock(i.next(), t)) {
i.remove();
return true;
}
@@ -212,10 +210,10 @@ public class Dispatcher {
/**
* @return true if the given block is good for the tentative move.
*/
- private boolean markMovedIfGoodBlock(DBlock block) {
+ private boolean markMovedIfGoodBlock(DBlock block, StorageType targetStorageType) {
synchronized (block) {
synchronized (movedBlocks) {
- if (isGoodBlockCandidate(source, target, block)) {
+ if (isGoodBlockCandidate(source, target, targetStorageType, block)) {
this.block = block;
if (chooseProxySource()) {
movedBlocks.put(block);
@@ -235,7 +233,7 @@ public class Dispatcher {
*
* @return true if a proxy is found; otherwise false
*/
- public boolean chooseProxySource() {
+ private boolean chooseProxySource() {
final DatanodeInfo targetDN = target.getDatanodeInfo();
// if node group is supported, first try add nodes in the same node group
if (cluster.isNodeGroupAware()) {
@@ -440,6 +438,18 @@ public class Dispatcher {
scheduledSize = 0L;
}
+ private PendingMove addPendingMove(DBlock block, final PendingMove pm) {
+ if (getDDatanode().addPendingBlock(pm)) {
+ if (pm.markMovedIfGoodBlock(block, getStorageType())) {
+ incScheduledSize(pm.block.getNumBytes());
+ return pm;
+ } else {
+ getDDatanode().removePendingBlock(pm);
+ }
+ }
+ return null;
+ }
+
/** @return the name for display */
String getDisplayName() {
return datanode + ":" + storageType;
@@ -599,8 +609,11 @@ public class Dispatcher {
/** Decide if the given block is a good candidate to move or not */
private boolean isGoodBlockCandidate(DBlock block) {
+ // source and target must have the same storage type
+ final StorageType sourceStorageType = getStorageType();
for (Task t : tasks) {
- if (Dispatcher.this.isGoodBlockCandidate(this, t.target, block)) {
+ if (Dispatcher.this.isGoodBlockCandidate(this, t.target,
+ sourceStorageType, block)) {
return true;
}
}
@@ -620,11 +633,9 @@ public class Dispatcher {
for (Iterator<Task> i = tasks.iterator(); i.hasNext();) {
final Task task = i.next();
final DDatanode target = task.target.getDDatanode();
- PendingMove pendingBlock = new PendingMove();
+ final PendingMove pendingBlock = new PendingMove(this, task.target);
if (target.addPendingBlock(pendingBlock)) {
// target is not busy, so do a tentative block allocation
- pendingBlock.source = this;
- pendingBlock.target = task.target;
if (pendingBlock.chooseBlockAndProxy()) {
long blockSize = pendingBlock.block.getNumBytes();
incScheduledSize(-blockSize);
@@ -641,6 +652,11 @@ public class Dispatcher {
}
return null;
}
+
+ /** Add a pending move */
+ public PendingMove addPendingMove(DBlock block, StorageGroup target) {
+ return target.addPendingMove(block, new PendingMove(this, target));
+ }
/** Iterate all source's blocks to remove moved ones */
private void removeMovedBlocks() {
@@ -901,12 +917,6 @@ public class Dispatcher {
}
}
- private boolean isGoodBlockCandidate(StorageGroup source, StorageGroup target,
- DBlock block) {
- // match source and target storage type
- return isGoodBlockCandidate(source, target, source.getStorageType(), block);
- }
-
/**
* Decide if the block is a good candidate to be moved from source to target.
* A block is a good candidate if
@@ -914,7 +924,7 @@ public class Dispatcher {
* 2. the block does not have a replica on the target;
* 3. doing the move does not reduce the number of racks that the block has
*/
- public boolean isGoodBlockCandidate(StorageGroup source, StorageGroup target,
+ private boolean isGoodBlockCandidate(StorageGroup source, StorageGroup target,
StorageType targetStorageType, DBlock block) {
if (target.storageType != targetStorageType) {
return false;
http://git-wip-us.apache.org/repos/asf/hadoop/blob/8ea20b53/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/balancer/Matcher.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/balancer/Matcher.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/balancer/Matcher.java
index 54febc6..f8d0071 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/balancer/Matcher.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/balancer/Matcher.java
@@ -31,6 +31,11 @@ public interface Matcher {
public boolean match(NetworkTopology cluster, Node left, Node right) {
return cluster.isOnSameNodeGroup(left, right);
}
+
+ @Override
+ public String toString() {
+ return "SAME_NODE_GROUP";
+ }
};
/** Match datanodes in the same rack. */
@@ -39,6 +44,11 @@ public interface Matcher {
public boolean match(NetworkTopology cluster, Node left, Node right) {
return cluster.isOnSameRack(left, right);
}
+
+ @Override
+ public String toString() {
+ return "SAME_RACK";
+ }
};
/** Match any datanode with any other datanode. */
@@ -47,5 +57,10 @@ public interface Matcher {
public boolean match(NetworkTopology cluster, Node left, Node right) {
return left != right;
}
+
+ @Override
+ public String toString() {
+ return "ANY_OTHER";
+ }
};
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/hadoop/blob/8ea20b53/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 424998c..4dbe1d3 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
@@ -50,7 +50,7 @@ import java.util.*;
public class Mover {
static final Log LOG = LogFactory.getLog(Mover.class);
- private static final Path MOVER_ID_PATH = new Path("/system/mover.id");
+ static final Path MOVER_ID_PATH = new Path("/system/mover.id");
private static class StorageMap {
private final StorageGroupMap<Source> sources
@@ -111,22 +111,25 @@ public class Mover {
this.storages = new StorageMap();
this.blockStoragePolicies = BlockStoragePolicy.readBlockStorageSuite(conf);
}
-
- private ExitStatus run() {
- try {
- final List<DatanodeStorageReport> reports = dispatcher.init();
- for(DatanodeStorageReport r : reports) {
- final DDatanode dn = dispatcher.newDatanode(r.getDatanodeInfo());
- for(StorageType t : StorageType.asList()) {
- final long maxRemaining = getMaxRemaining(r, t);
- if (maxRemaining > 0L) {
- final Source source = dn.addSource(t, Long.MAX_VALUE, dispatcher);
- final StorageGroup target = dn.addTarget(t, maxRemaining);
- storages.add(source, target);
- }
+
+ void init() throws IOException {
+ final List<DatanodeStorageReport> reports = dispatcher.init();
+ for(DatanodeStorageReport r : reports) {
+ final DDatanode dn = dispatcher.newDatanode(r.getDatanodeInfo());
+ for(StorageType t : StorageType.asList()) {
+ final long maxRemaining = getMaxRemaining(r, t);
+ if (maxRemaining > 0L) {
+ final Source source = dn.addSource(t, Long.MAX_VALUE, dispatcher);
+ final StorageGroup target = dn.addTarget(t, maxRemaining);
+ storages.add(source, target);
}
}
+ }
+ }
+ private ExitStatus run() {
+ try {
+ init();
new Processor().processNamespace();
return ExitStatus.IN_PROGRESS;
@@ -141,6 +144,14 @@ public class Mover {
}
}
+ DBlock newDBlock(Block block, List<MLocation> locations) {
+ final DBlock db = new DBlock(block);
+ for(MLocation ml : locations) {
+ db.addLocation(storages.getTarget(ml));
+ }
+ return db;
+ }
+
private static long getMaxRemaining(DatanodeStorageReport report, StorageType t) {
long max = 0L;
for(StorageReport r : report.getStorageReports()) {
@@ -169,11 +180,11 @@ public class Mover {
return sb.toString();
}
- private class Processor {
+ class Processor {
private final DFSClient dfs;
private final List<String> snapshottableDirs = new ArrayList<String>();
- private Processor() {
+ Processor() {
dfs = dispatcher.getDistributedFileSystem().getClient();
}
@@ -290,15 +301,11 @@ public class Mover {
}
}
}
-
+
void scheduleMoves4Block(StorageTypeDiff diff, LocatedBlock lb) {
final List<MLocation> locations = MLocation.toLocations(lb);
Collections.shuffle(locations);
-
- final DBlock db = new DBlock(lb.getBlock().getLocalBlock());
- for(MLocation ml : locations) {
- db.addLocation(storages.getTarget(ml));
- }
+ final DBlock db = newDBlock(lb.getBlock().getLocalBlock(), locations);
for(final Iterator<StorageType> i = diff.existing.iterator(); i.hasNext(); ) {
final StorageType t = i.next();
@@ -310,12 +317,18 @@ public class Mover {
if (scheduleMoveReplica(db, ml, source, diff.expected)) {
i.remove();
j.remove();
+ return;
}
}
}
}
}
+ boolean scheduleMoveReplica(DBlock db, MLocation ml,
+ List<StorageType> targetTypes) {
+ return scheduleMoveReplica(db, ml, storages.getSource(ml), targetTypes);
+ }
+
boolean scheduleMoveReplica(DBlock db, MLocation ml, Source source,
List<StorageType> targetTypes) {
if (dispatcher.getCluster().isNodeGroupAware()) {
@@ -341,12 +354,10 @@ public class Mover {
for(final Iterator<StorageType> i = targetTypes.iterator(); i.hasNext(); ) {
final StorageType t = i.next();
for(StorageGroup target : storages.getTargetStorages(t)) {
- if (matcher.match(cluster, ml.datanode, target.getDatanodeInfo())
- && dispatcher.isGoodBlockCandidate(source, target, t, db)) {
- final PendingMove pm = dispatcher.new PendingMove(db, source, target);
- if (pm.chooseProxySource()) {
+ if (matcher.match(cluster, ml.datanode, target.getDatanodeInfo())) {
+ final PendingMove pm = source.addPendingMove(db, target);
+ if (pm != null) {
i.remove();
- target.incScheduledSize(ml.size);
dispatcher.executePendingMove(pm);
return true;
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/8ea20b53/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 f2c9fb1..ff5e995 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
@@ -20,7 +20,6 @@ package org.apache.hadoop.hdfs;
import static org.apache.hadoop.hdfs.BlockStoragePolicy.ID_UNSPECIFIED;
import java.io.FileNotFoundException;
-import java.util.ArrayList;
import java.util.Arrays;
import java.util.EnumSet;
import java.util.HashMap;
@@ -62,20 +61,6 @@ public class TestBlockStoragePolicy {
static final byte WARM = (byte) 8;
static final byte HOT = (byte) 12;
- static final List<List<StorageType>> chosens = new ArrayList<List<StorageType>>();
- static {
- chosens.add(Arrays.<StorageType>asList());
- chosens.add(Arrays.asList(StorageType.DISK));
- chosens.add(Arrays.asList(StorageType.ARCHIVE));
- chosens.add(Arrays.asList(StorageType.DISK, StorageType.DISK));
- chosens.add(Arrays.asList(StorageType.DISK, StorageType.ARCHIVE));
- chosens.add(Arrays.asList(StorageType.ARCHIVE, StorageType.ARCHIVE));
- chosens.add(Arrays.asList(StorageType.DISK, StorageType.DISK, StorageType.DISK));
- chosens.add(Arrays.asList(StorageType.DISK, StorageType.DISK, StorageType.ARCHIVE));
- chosens.add(Arrays.asList(StorageType.DISK, StorageType.ARCHIVE, StorageType.ARCHIVE));
- chosens.add(Arrays.asList(StorageType.ARCHIVE, StorageType.ARCHIVE, StorageType.ARCHIVE));
- }
-
@Test
public void testDefaultPolicies() {
final Map<Byte, String> expectedPolicyStrings = new HashMap<Byte, String>();
@@ -126,6 +111,17 @@ public class TestBlockStoragePolicy {
}
}
+ static StorageType[] newStorageTypes(int nDisk, int nArchive) {
+ final StorageType[] t = new StorageType[nDisk + nArchive];
+ Arrays.fill(t, 0, nDisk, StorageType.DISK);
+ Arrays.fill(t, nDisk, t.length, StorageType.ARCHIVE);
+ return t;
+ }
+
+ static List<StorageType> asList(int nDisk, int nArchive) {
+ return Arrays.asList(newStorageTypes(nDisk, nArchive));
+ }
+
static void assertStorageType(List<StorageType> computed, short replication,
StorageType... answers) {
Assert.assertEquals(replication, computed.size());
@@ -369,10 +365,14 @@ public class TestBlockStoragePolicy {
final BlockStoragePolicy cold = POLICY_SUITE.getPolicy(COLD);
final short replication = 3;
- for(List<StorageType> c : chosens) {
- method.checkChooseStorageTypes(hot, replication, c);
- method.checkChooseStorageTypes(warm, replication, c);
- method.checkChooseStorageTypes(cold, replication, c);
+ for(int n = 0; n <= 3; n++) {
+ for(int d = 0; d <= n; d++) {
+ final int a = n - d;
+ final List<StorageType> chosen = asList(d, a);
+ method.checkChooseStorageTypes(hot, replication, chosen);
+ method.checkChooseStorageTypes(warm, replication, chosen);
+ method.checkChooseStorageTypes(cold, replication, chosen);
+ }
}
}
@@ -714,6 +714,47 @@ public class TestBlockStoragePolicy {
Assert.assertArrayEquals(expected, computed);
}
+ @Test
+ public void testChooseExcess() {
+ final BlockStoragePolicy hot = POLICY_SUITE.getPolicy(HOT);
+ final BlockStoragePolicy warm = POLICY_SUITE.getPolicy(WARM);
+ final BlockStoragePolicy cold = POLICY_SUITE.getPolicy(COLD);
+
+ final short replication = 3;
+ for(int n = 0; n <= 6; n++) {
+ for(int d = 0; d <= n; d++) {
+ final int a = n - d;
+ final List<StorageType> chosen = asList(d, a);
+ {
+ final int nDisk = Math.max(0, d - replication);
+ final int nArchive = a;
+ final StorageType[] expected = newStorageTypes(nDisk, nArchive);
+ checkChooseExcess(hot, replication, chosen, expected);
+ }
+
+ {
+ final int nDisk = Math.max(0, d - 1);
+ final int nArchive = Math.max(0, a - replication + 1);
+ final StorageType[] expected = newStorageTypes(nDisk, nArchive);
+ checkChooseExcess(warm, replication, chosen, expected);
+ }
+
+ {
+ final int nDisk = d;
+ final int nArchive = Math.max(0, a - replication );
+ final StorageType[] expected = newStorageTypes(nDisk, nArchive);
+ checkChooseExcess(cold, replication, chosen, expected);
+ }
+ }
+ }
+ }
+
+ static void checkChooseExcess(BlockStoragePolicy p, short replication,
+ List<StorageType> chosen, StorageType... expected) {
+ final List<StorageType> types = p.chooseExcess(replication, chosen);
+ assertStorageTypes(types, expected);
+ }
+
private void checkDirectoryListing(HdfsFileStatus[] stats, byte... policies) {
Assert.assertEquals(stats.length, policies.length);
for (int i = 0; i < stats.length; i++) {
http://git-wip-us.apache.org/repos/asf/hadoop/blob/8ea20b53/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/mover/TestMover.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/mover/TestMover.java b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/mover/TestMover.java
new file mode 100644
index 0000000..da913e7
--- /dev/null
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/mover/TestMover.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.hadoop.hdfs.server.mover;
+
+import java.io.IOException;
+import java.net.URI;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.List;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FSDataOutputStream;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hdfs.DFSUtil;
+import org.apache.hadoop.hdfs.DistributedFileSystem;
+import org.apache.hadoop.hdfs.HdfsConfiguration;
+import org.apache.hadoop.hdfs.MiniDFSCluster;
+import org.apache.hadoop.hdfs.StorageType;
+import org.apache.hadoop.hdfs.protocol.LocatedBlock;
+import org.apache.hadoop.hdfs.server.balancer.Dispatcher.DBlock;
+import org.apache.hadoop.hdfs.server.balancer.NameNodeConnector;
+import org.apache.hadoop.hdfs.server.mover.Mover.MLocation;
+import org.junit.Assert;
+import org.junit.Test;
+
+public class TestMover {
+ static Mover newMover(Configuration conf) throws IOException {
+ final Collection<URI> namenodes = DFSUtil.getNsServiceRpcUris(conf);
+ Assert.assertEquals(1, namenodes.size());
+
+ final List<NameNodeConnector> nncs = NameNodeConnector.newNameNodeConnectors(
+ namenodes, Mover.class.getSimpleName(), Mover.MOVER_ID_PATH, conf);
+ return new Mover(nncs.get(0), conf);
+ }
+
+ @Test
+ public void testScheduleSameBlock() throws IOException {
+ final Configuration conf = new HdfsConfiguration();
+ final MiniDFSCluster cluster = new MiniDFSCluster.Builder(conf)
+ .numDataNodes(4).build();
+ try {
+ cluster.waitActive();
+ final DistributedFileSystem dfs = cluster.getFileSystem();
+ final String file = "/testScheduleSameBlock/file";
+
+ {
+ final FSDataOutputStream out = dfs.create(new Path(file));
+ out.writeChars("testScheduleSameBlock");
+ out.close();
+ }
+
+ final Mover mover = newMover(conf);
+ mover.init();
+ final Mover.Processor processor = mover.new Processor();
+
+ final LocatedBlock lb = dfs.getClient().getLocatedBlocks(file, 0).get(0);
+ final List<MLocation> locations = MLocation.toLocations(lb);
+ final MLocation ml = locations.get(0);
+ final DBlock db = mover.newDBlock(lb.getBlock().getLocalBlock(), locations);
+
+ final List<StorageType> storageTypes = new ArrayList<StorageType>(
+ Arrays.asList(StorageType.DEFAULT, StorageType.DEFAULT));
+ Assert.assertTrue(processor.scheduleMoveReplica(db, ml, storageTypes));
+ Assert.assertFalse(processor.scheduleMoveReplica(db, ml, storageTypes));
+ } finally {
+ cluster.shutdown();
+ }
+ }
+}