big-data/hadoop
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

HDFS-10988. Refactor TestBalancerBandwidth. Contributed by Brahma Reddy Battula

Browse Source
This commit is contained in:
Mingliang Liu 2016-10-10 13:19:17 -07:00
parent 65912e4027
commit b963818621
1 changed files with 25 additions and 32 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

57
hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/TestBalancerBandwidth.java
View File

@ -24,13 +24,15 @@
import java.io.PrintStream; import java.io.PrintStream;
import java.nio.charset.Charset; import java.nio.charset.Charset;
import java.util.ArrayList; import java.util.ArrayList;
import java.util.concurrent.TimeoutException;
import com.google.common.base.Supplier;
import org.apache.commons.logging.Log; import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory; import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hdfs.protocol.ClientDatanodeProtocol;
import org.apache.hadoop.hdfs.server.datanode.DataNode; import org.apache.hadoop.hdfs.server.datanode.DataNode;
import org.apache.hadoop.hdfs.tools.DFSAdmin; import org.apache.hadoop.hdfs.tools.DFSAdmin;
import org.apache.hadoop.test.GenericTestUtils;
import org.junit.Test; import org.junit.Test;
/** /**
@ -54,9 +56,8 @@ public void testBalancerBandwidth() throws Exception {
DEFAULT_BANDWIDTH); DEFAULT_BANDWIDTH);
/* Create and start cluster */ /* Create and start cluster */
MiniDFSCluster cluster = try (MiniDFSCluster cluster = new MiniDFSCluster.Builder(conf)
new MiniDFSCluster.Builder(conf).numDataNodes(NUM_OF_DATANODES).build(); .numDataNodes(NUM_OF_DATANODES).build()) {
try {
cluster.waitActive(); cluster.waitActive();
DistributedFileSystem fs = cluster.getFileSystem(); DistributedFileSystem fs = cluster.getFileSystem();
@ -65,12 +66,6 @@ public void testBalancerBandwidth() throws Exception {
// Ensure value from the configuration is reflected in the datanodes. // Ensure value from the configuration is reflected in the datanodes.
assertEquals(DEFAULT_BANDWIDTH, (long) datanodes.get(0).getBalancerBandwidth()); assertEquals(DEFAULT_BANDWIDTH, (long) datanodes.get(0).getBalancerBandwidth());
assertEquals(DEFAULT_BANDWIDTH, (long) datanodes.get(1).getBalancerBandwidth()); assertEquals(DEFAULT_BANDWIDTH, (long) datanodes.get(1).getBalancerBandwidth());
ClientDatanodeProtocol dn1Proxy = DFSUtilClient
.createClientDatanodeProtocolProxy(datanodes.get(0).getDatanodeId(),
conf, 60000, false);
ClientDatanodeProtocol dn2Proxy = DFSUtilClient
.createClientDatanodeProtocolProxy(datanodes.get(1).getDatanodeId(),
conf, 60000, false);
DFSAdmin admin = new DFSAdmin(conf); DFSAdmin admin = new DFSAdmin(conf);
String dn1Address = datanodes.get(0).ipcServer.getListenerAddress() String dn1Address = datanodes.get(0).ipcServer.getListenerAddress()
.getHostName() + ":" + datanodes.get(0).getIpcPort(); .getHostName() + ":" + datanodes.get(0).getIpcPort();
@ -79,51 +74,49 @@ public void testBalancerBandwidth() throws Exception {
// verifies the dfsadmin command execution // verifies the dfsadmin command execution
String[] args = new String[] { "-getBalancerBandwidth", dn1Address }; String[] args = new String[] { "-getBalancerBandwidth", dn1Address };
runGetBalancerBandwidthCmd(admin, args, dn1Proxy, DEFAULT_BANDWIDTH); runGetBalancerBandwidthCmd(admin, args, DEFAULT_BANDWIDTH);
args = new String[] { "-getBalancerBandwidth", dn2Address }; args = new String[] { "-getBalancerBandwidth", dn2Address };
runGetBalancerBandwidthCmd(admin, args, dn2Proxy, DEFAULT_BANDWIDTH); runGetBalancerBandwidthCmd(admin, args, DEFAULT_BANDWIDTH);
// Dynamically change balancer bandwidth and ensure the updated value // Dynamically change balancer bandwidth and ensure the updated value
// is reflected on the datanodes. // is reflected on the datanodes.
long newBandwidth = 12 * DEFAULT_BANDWIDTH; // 12M bps long newBandwidth = 12 * DEFAULT_BANDWIDTH; // 12M bps
fs.setBalancerBandwidth(newBandwidth); fs.setBalancerBandwidth(newBandwidth);
verifyBalancerBandwidth(datanodes, newBandwidth);
// Give it a few seconds to propogate new the value to the datanodes.
try {
Thread.sleep(5000);
} catch (Exception e) {}
assertEquals(newBandwidth, (long) datanodes.get(0).getBalancerBandwidth());
assertEquals(newBandwidth, (long) datanodes.get(1).getBalancerBandwidth());
// verifies the dfsadmin command execution // verifies the dfsadmin command execution
args = new String[] { "-getBalancerBandwidth", dn1Address }; args = new String[] { "-getBalancerBandwidth", dn1Address };
runGetBalancerBandwidthCmd(admin, args, dn1Proxy, newBandwidth); runGetBalancerBandwidthCmd(admin, args, newBandwidth);
args = new String[] { "-getBalancerBandwidth", dn2Address }; args = new String[] { "-getBalancerBandwidth", dn2Address };
runGetBalancerBandwidthCmd(admin, args, dn2Proxy, newBandwidth); runGetBalancerBandwidthCmd(admin, args, newBandwidth);
// Dynamically change balancer bandwidth to 0. Balancer bandwidth on the // Dynamically change balancer bandwidth to 0. Balancer bandwidth on the
// datanodes should remain as it was. // datanodes should remain as it was.
fs.setBalancerBandwidth(0); fs.setBalancerBandwidth(0);
// Give it a few seconds to propogate new the value to the datanodes. verifyBalancerBandwidth(datanodes, newBandwidth);
try {
Thread.sleep(5000);
} catch (Exception e) {}
assertEquals(newBandwidth, (long) datanodes.get(0).getBalancerBandwidth());
assertEquals(newBandwidth, (long) datanodes.get(1).getBalancerBandwidth());
// verifies the dfsadmin command execution // verifies the dfsadmin command execution
args = new String[] { "-getBalancerBandwidth", dn1Address }; args = new String[] { "-getBalancerBandwidth", dn1Address };
runGetBalancerBandwidthCmd(admin, args, dn1Proxy, newBandwidth); runGetBalancerBandwidthCmd(admin, args, newBandwidth);
args = new String[] { "-getBalancerBandwidth", dn2Address }; args = new String[] { "-getBalancerBandwidth", dn2Address };
runGetBalancerBandwidthCmd(admin, args, dn2Proxy, newBandwidth); runGetBalancerBandwidthCmd(admin, args, newBandwidth);
} finally {
cluster.shutdown();
} }
} }
private void verifyBalancerBandwidth(final ArrayList<DataNode> datanodes,
final long newBandwidth) throws TimeoutException, InterruptedException {
GenericTestUtils.waitFor(new Supplier<Boolean>() {
@Override
public Boolean get() {
return (long) datanodes.get(0).getBalancerBandwidth() == newBandwidth
&& (long) datanodes.get(1).getBalancerBandwidth() == newBandwidth;
}
}, 100, 60 * 1000);
}
private void runGetBalancerBandwidthCmd(DFSAdmin admin, String[] args, private void runGetBalancerBandwidthCmd(DFSAdmin admin, String[] args,
ClientDatanodeProtocol proxy, long expectedBandwidth) throws Exception { long expectedBandwidth) throws Exception {
PrintStream initialStdOut = System.out; PrintStream initialStdOut = System.out;
outContent.reset(); outContent.reset();
try { try {