| /** |
| * 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.datanode; |
| |
| import org.apache.commons.logging.Log; |
| import org.apache.commons.logging.LogFactory; |
| import org.apache.commons.logging.impl.Log4JLogger; |
| import org.apache.hadoop.conf.Configuration; |
| import org.apache.hadoop.fs.FSDataOutputStream; |
| import org.apache.hadoop.fs.Path; |
| import org.apache.hadoop.hdfs.AppendTestUtil; |
| import org.apache.hadoop.hdfs.DFSConfigKeys; |
| import org.apache.hadoop.hdfs.DFSTestUtil; |
| import org.apache.hadoop.hdfs.DistributedFileSystem; |
| import org.apache.hadoop.hdfs.MiniDFSCluster; |
| import org.apache.hadoop.hdfs.protocol.Block; |
| import org.apache.hadoop.hdfs.protocol.BlockListAsLongs; |
| import org.apache.hadoop.hdfs.protocol.ExtendedBlock; |
| import org.apache.hadoop.hdfs.protocol.LocatedBlock; |
| import org.apache.hadoop.hdfs.server.blockmanagement.BlockManagerTestUtil; |
| import org.apache.hadoop.hdfs.server.common.HdfsServerConstants; |
| import org.apache.hadoop.hdfs.server.namenode.FSNamesystem; |
| import org.apache.hadoop.hdfs.server.namenode.NameNode; |
| import org.apache.hadoop.hdfs.server.protocol.DatanodeCommand; |
| import org.apache.hadoop.hdfs.server.protocol.DatanodeProtocol; |
| import org.apache.hadoop.hdfs.server.protocol.DatanodeRegistration; |
| import org.apache.hadoop.io.IOUtils; |
| import org.apache.hadoop.hdfs.server.protocol.StorageBlockReport; |
| import org.apache.hadoop.test.GenericTestUtils; |
| import org.apache.hadoop.test.GenericTestUtils.DelayAnswer; |
| import org.apache.log4j.Level; |
| import org.junit.After; |
| import static org.junit.Assert.assertEquals; |
| import static org.junit.Assert.assertTrue; |
| import org.junit.Before; |
| import org.junit.Test; |
| import org.mockito.Mockito; |
| import org.mockito.invocation.InvocationOnMock; |
| |
| import java.io.File; |
| import java.io.FilenameFilter; |
| import java.io.IOException; |
| import java.util.ArrayList; |
| import java.util.List; |
| import java.util.Random; |
| import java.util.concurrent.CountDownLatch; |
| |
| /** |
| * This test simulates a variety of situations when blocks are being |
| * intentionally orrupted, unexpectedly modified, and so on before a block |
| * report is happening |
| */ |
| public class TestBlockReport { |
| public static final Log LOG = LogFactory.getLog(TestBlockReport.class); |
| |
| private static short REPL_FACTOR = 1; |
| private static final int RAND_LIMIT = 2000; |
| private static final long DN_RESCAN_INTERVAL = 5000; |
| private static final long DN_RESCAN_EXTRA_WAIT = 2 * DN_RESCAN_INTERVAL; |
| private static final int DN_N0 = 0; |
| private static final int FILE_START = 0; |
| |
| static final int BLOCK_SIZE = 1024; |
| static final int NUM_BLOCKS = 10; |
| static final int FILE_SIZE = NUM_BLOCKS * BLOCK_SIZE + 1; |
| static String bpid; |
| |
| private MiniDFSCluster cluster; |
| private DistributedFileSystem fs; |
| |
| Random rand = new Random(RAND_LIMIT); |
| |
| private static Configuration conf; |
| |
| static { |
| initLoggers(); |
| resetConfiguration(); |
| } |
| |
| @Before |
| public void startUpCluster() throws IOException { |
| REPL_FACTOR = 1; //Reset if case a test has modified the value |
| cluster = new MiniDFSCluster.Builder(conf).numDataNodes(REPL_FACTOR).build(); |
| fs = (DistributedFileSystem) cluster.getFileSystem(); |
| bpid = cluster.getNamesystem().getBlockPoolId(); |
| } |
| |
| @After |
| public void shutDownCluster() throws IOException { |
| fs.close(); |
| cluster.shutdownDataNodes(); |
| cluster.shutdown(); |
| } |
| |
| /** |
| * Test write a file, verifies and closes it. Then the length of the blocks |
| * are messed up and BlockReport is forced. |
| * The modification of blocks' length has to be ignored |
| * |
| * @throws java.io.IOException on an error |
| */ |
| @Test |
| public void blockReport_01() throws IOException { |
| final String METHOD_NAME = GenericTestUtils.getMethodName(); |
| Path filePath = new Path("/" + METHOD_NAME + ".dat"); |
| |
| ArrayList<Block> blocks = prepareForRide(filePath, METHOD_NAME, FILE_SIZE); |
| |
| if(LOG.isDebugEnabled()) { |
| LOG.debug("Number of blocks allocated " + blocks.size()); |
| } |
| long[] oldLengths = new long[blocks.size()]; |
| int tempLen; |
| for (int i = 0; i < blocks.size(); i++) { |
| Block b = blocks.get(i); |
| if(LOG.isDebugEnabled()) { |
| LOG.debug("Block " + b.getBlockName() + " before\t" + "Size " + |
| b.getNumBytes()); |
| } |
| oldLengths[i] = b.getNumBytes(); |
| if(LOG.isDebugEnabled()) { |
| LOG.debug("Setting new length"); |
| } |
| tempLen = rand.nextInt(BLOCK_SIZE); |
| b.set(b.getBlockId(), tempLen, b.getGenerationStamp()); |
| if(LOG.isDebugEnabled()) { |
| LOG.debug("Block " + b.getBlockName() + " after\t " + "Size " + |
| b.getNumBytes()); |
| } |
| } |
| // all blocks belong to the same file, hence same BP |
| DataNode dn = cluster.getDataNodes().get(DN_N0); |
| String poolId = cluster.getNamesystem().getBlockPoolId(); |
| DatanodeRegistration dnR = dn.getDNRegistrationForBP(poolId); |
| StorageBlockReport[] report = { new StorageBlockReport(dnR.getStorageID(), |
| new BlockListAsLongs(blocks, null).getBlockListAsLongs()) }; |
| cluster.getNameNodeRpc().blockReport(dnR, poolId, report); |
| |
| List<LocatedBlock> blocksAfterReport = |
| DFSTestUtil.getAllBlocks(fs.open(filePath)); |
| |
| if(LOG.isDebugEnabled()) { |
| LOG.debug("After mods: Number of blocks allocated " + |
| blocksAfterReport.size()); |
| } |
| |
| for (int i = 0; i < blocksAfterReport.size(); i++) { |
| ExtendedBlock b = blocksAfterReport.get(i).getBlock(); |
| assertEquals("Length of " + i + "th block is incorrect", |
| oldLengths[i], b.getNumBytes()); |
| } |
| } |
| |
| /** |
| * Test write a file, verifies and closes it. Then a couple of random blocks |
| * is removed and BlockReport is forced; the FSNamesystem is pushed to |
| * recalculate required DN's activities such as replications and so on. |
| * The number of missing and under-replicated blocks should be the same in |
| * case of a single-DN cluster. |
| * |
| * @throws IOException in case of errors |
| */ |
| @Test |
| public void blockReport_02() throws IOException { |
| final String METHOD_NAME = GenericTestUtils.getMethodName(); |
| LOG.info("Running test " + METHOD_NAME); |
| |
| Path filePath = new Path("/" + METHOD_NAME + ".dat"); |
| DFSTestUtil.createFile(fs, filePath, |
| FILE_SIZE, REPL_FACTOR, rand.nextLong()); |
| |
| // mock around with newly created blocks and delete some |
| File dataDir = new File(cluster.getDataDirectory()); |
| assertTrue(dataDir.isDirectory()); |
| |
| List<ExtendedBlock> blocks2Remove = new ArrayList<ExtendedBlock>(); |
| List<Integer> removedIndex = new ArrayList<Integer>(); |
| List<LocatedBlock> lBlocks = |
| cluster.getNameNodeRpc().getBlockLocations( |
| filePath.toString(), FILE_START, |
| FILE_SIZE).getLocatedBlocks(); |
| |
| while (removedIndex.size() != 2) { |
| int newRemoveIndex = rand.nextInt(lBlocks.size()); |
| if (!removedIndex.contains(newRemoveIndex)) |
| removedIndex.add(newRemoveIndex); |
| } |
| |
| for (Integer aRemovedIndex : removedIndex) { |
| blocks2Remove.add(lBlocks.get(aRemovedIndex).getBlock()); |
| } |
| ArrayList<Block> blocks = locatedToBlocks(lBlocks, removedIndex); |
| |
| if(LOG.isDebugEnabled()) { |
| LOG.debug("Number of blocks allocated " + lBlocks.size()); |
| } |
| |
| for (ExtendedBlock b : blocks2Remove) { |
| if(LOG.isDebugEnabled()) { |
| LOG.debug("Removing the block " + b.getBlockName()); |
| } |
| for (File f : findAllFiles(dataDir, |
| new MyFileFilter(b.getBlockName(), true))) { |
| cluster.getDataNodes().get(DN_N0).getFSDataset().unfinalizeBlock(b); |
| if (!f.delete()) |
| LOG.warn("Couldn't delete " + b.getBlockName()); |
| } |
| } |
| |
| waitTil(DN_RESCAN_EXTRA_WAIT); |
| |
| // all blocks belong to the same file, hence same BP |
| DataNode dn = cluster.getDataNodes().get(DN_N0); |
| String poolId = cluster.getNamesystem().getBlockPoolId(); |
| DatanodeRegistration dnR = dn.getDNRegistrationForBP(poolId); |
| StorageBlockReport[] report = { new StorageBlockReport(dnR.getStorageID(), |
| new BlockListAsLongs(blocks, null).getBlockListAsLongs()) }; |
| cluster.getNameNodeRpc().blockReport(dnR, poolId, report); |
| |
| BlockManagerTestUtil.getComputedDatanodeWork(cluster.getNamesystem() |
| .getBlockManager()); |
| |
| printStats(); |
| |
| assertEquals("Wrong number of MissingBlocks is found", |
| blocks2Remove.size(), cluster.getNamesystem().getMissingBlocksCount()); |
| assertEquals("Wrong number of UnderReplicatedBlocks is found", |
| blocks2Remove.size(), cluster.getNamesystem().getUnderReplicatedBlocks()); |
| } |
| |
| |
| /** |
| * Test writes a file and closes it. Then test finds a block |
| * and changes its GS to be < of original one. |
| * New empty block is added to the list of blocks. |
| * Block report is forced and the check for # of corrupted blocks is performed. |
| * |
| * @throws IOException in case of an error |
| */ |
| @Test |
| public void blockReport_03() throws IOException { |
| final String METHOD_NAME = GenericTestUtils.getMethodName(); |
| Path filePath = new Path("/" + METHOD_NAME + ".dat"); |
| |
| ArrayList<Block> blocks = |
| prepareForRide(filePath, METHOD_NAME, FILE_SIZE); |
| |
| // The block with modified GS won't be found. Has to be deleted |
| blocks.get(0).setGenerationStamp(rand.nextLong()); |
| // This new block is unknown to NN and will be mark for deletion. |
| blocks.add(new Block()); |
| |
| // all blocks belong to the same file, hence same BP |
| DataNode dn = cluster.getDataNodes().get(DN_N0); |
| String poolId = cluster.getNamesystem().getBlockPoolId(); |
| DatanodeRegistration dnR = dn.getDNRegistrationForBP(poolId); |
| StorageBlockReport[] report = { new StorageBlockReport(dnR.getStorageID(), |
| new BlockListAsLongs(blocks, null).getBlockListAsLongs()) }; |
| DatanodeCommand dnCmd = |
| cluster.getNameNodeRpc().blockReport(dnR, poolId, report); |
| if(LOG.isDebugEnabled()) { |
| LOG.debug("Got the command: " + dnCmd); |
| } |
| printStats(); |
| |
| assertEquals("Wrong number of CorruptedReplica+PendingDeletion " + |
| "blocks is found", 2, |
| cluster.getNamesystem().getCorruptReplicaBlocks() + |
| cluster.getNamesystem().getPendingDeletionBlocks()); |
| } |
| |
| /** |
| * This test isn't a representative case for BlockReport |
| * The empty method is going to be left here to keep the naming |
| * of the test plan in synch with the actual implementation |
| */ |
| public void blockReport_04() { |
| } |
| |
| // Client requests new block from NN. The test corrupts this very block |
| // and forces new block report. |
| // The test case isn't specific for BlockReport because it relies on |
| // BlockScanner which is out of scope of this test |
| // Keeping the name to be in synch with the test plan |
| // |
| public void blockReport_05() { |
| } |
| |
| /** |
| * Test creates a file and closes it. |
| * The second datanode is started in the cluster. |
| * As soon as the replication process is completed test runs |
| * Block report and checks that no underreplicated blocks are left |
| * |
| * @throws IOException in case of an error |
| */ |
| @Test |
| public void blockReport_06() throws IOException { |
| final String METHOD_NAME = GenericTestUtils.getMethodName(); |
| Path filePath = new Path("/" + METHOD_NAME + ".dat"); |
| final int DN_N1 = DN_N0 + 1; |
| |
| ArrayList<Block> blocks = writeFile(METHOD_NAME, FILE_SIZE, filePath); |
| startDNandWait(filePath, true); |
| |
| // all blocks belong to the same file, hence same BP |
| DataNode dn = cluster.getDataNodes().get(DN_N1); |
| String poolId = cluster.getNamesystem().getBlockPoolId(); |
| DatanodeRegistration dnR = dn.getDNRegistrationForBP(poolId); |
| StorageBlockReport[] report = { new StorageBlockReport(dnR.getStorageID(), |
| new BlockListAsLongs(blocks, null).getBlockListAsLongs()) }; |
| cluster.getNameNodeRpc().blockReport(dnR, poolId, report); |
| printStats(); |
| assertEquals("Wrong number of PendingReplication Blocks", |
| 0, cluster.getNamesystem().getUnderReplicatedBlocks()); |
| } |
| |
| /** |
| * Similar to BlockReport_03() but works with two DNs |
| * Test writes a file and closes it. |
| * The second datanode is started in the cluster. |
| * As soon as the replication process is completed test finds a block from |
| * the second DN and sets its GS to be < of original one. |
| * Block report is forced and the check for # of currupted blocks is performed. |
| * Another block is chosen and its length is set to a lesser than original. |
| * A check for another corrupted block is performed after yet another |
| * BlockReport |
| * |
| * @throws IOException in case of an error |
| */ |
| @Test |
| // Currently this test is failing as expected 'cause the correct behavior is |
| // not yet implemented (9/15/09) |
| public void blockReport_07() throws IOException { |
| final String METHOD_NAME = GenericTestUtils.getMethodName(); |
| Path filePath = new Path("/" + METHOD_NAME + ".dat"); |
| final int DN_N1 = DN_N0 + 1; |
| |
| // write file and start second node to be "older" than the original |
| ArrayList<Block> blocks = writeFile(METHOD_NAME, FILE_SIZE, filePath); |
| startDNandWait(filePath, true); |
| |
| int randIndex = rand.nextInt(blocks.size()); |
| // Get a block and screw its GS |
| Block corruptedBlock = blocks.get(randIndex); |
| String secondNode = cluster.getDataNodes().get(DN_N1).getStorageId(); |
| if(LOG.isDebugEnabled()) { |
| LOG.debug("Working with " + secondNode); |
| LOG.debug("BlockGS before " + blocks.get(randIndex).getGenerationStamp()); |
| } |
| corruptBlockGS(corruptedBlock); |
| if(LOG.isDebugEnabled()) { |
| LOG.debug("BlockGS after " + blocks.get(randIndex).getGenerationStamp()); |
| LOG.debug("Done corrupting GS of " + corruptedBlock.getBlockName()); |
| } |
| // all blocks belong to the same file, hence same BP |
| DataNode dn = cluster.getDataNodes().get(DN_N1); |
| String poolId = cluster.getNamesystem().getBlockPoolId(); |
| DatanodeRegistration dnR = dn.getDNRegistrationForBP(poolId); |
| StorageBlockReport[] report = { new StorageBlockReport(dnR.getStorageID(), |
| new BlockListAsLongs(blocks, null).getBlockListAsLongs()) }; |
| cluster.getNameNodeRpc().blockReport(dnR, poolId, report); |
| printStats(); |
| assertEquals("Wrong number of Corrupted blocks", |
| 1, cluster.getNamesystem().getCorruptReplicaBlocks() + |
| // the following might have to be added into the equation if |
| // the same block could be in two different states at the same time |
| // and then the expected number of has to be changed to '2' |
| // cluster.getNamesystem().getPendingReplicationBlocks() + |
| cluster.getNamesystem().getPendingDeletionBlocks()); |
| |
| // Get another block and screw its length to be less than original |
| if (randIndex == 0) |
| randIndex++; |
| else |
| randIndex--; |
| corruptedBlock = blocks.get(randIndex); |
| corruptBlockLen(corruptedBlock); |
| if(LOG.isDebugEnabled()) { |
| LOG.debug("Done corrupting length of " + corruptedBlock.getBlockName()); |
| } |
| |
| report[0] = new StorageBlockReport(dnR.getStorageID(), |
| new BlockListAsLongs(blocks, null).getBlockListAsLongs()); |
| cluster.getNameNodeRpc().blockReport(dnR, poolId, report); |
| printStats(); |
| |
| assertEquals("Wrong number of Corrupted blocks", |
| 2, cluster.getNamesystem().getCorruptReplicaBlocks() + |
| cluster.getNamesystem().getPendingReplicationBlocks() + |
| cluster.getNamesystem().getPendingDeletionBlocks()); |
| |
| printStats(); |
| |
| } |
| |
| /** |
| * The test set the configuration parameters for a large block size and |
| * restarts initiated single-node cluster. |
| * Then it writes a file > block_size and closes it. |
| * The second datanode is started in the cluster. |
| * As soon as the replication process is started and at least one TEMPORARY |
| * replica is found test forces BlockReport process and checks |
| * if the TEMPORARY replica isn't reported on it. |
| * Eventually, the configuration is being restored into the original state. |
| * |
| * @throws IOException in case of an error |
| */ |
| @Test |
| public void blockReport_08() throws IOException { |
| final String METHOD_NAME = GenericTestUtils.getMethodName(); |
| Path filePath = new Path("/" + METHOD_NAME + ".dat"); |
| final int DN_N1 = DN_N0 + 1; |
| final int bytesChkSum = 1024 * 1000; |
| |
| conf.setInt(DFSConfigKeys.DFS_BYTES_PER_CHECKSUM_KEY, bytesChkSum); |
| conf.setLong(DFSConfigKeys.DFS_BLOCK_SIZE_KEY, 6 * bytesChkSum); |
| shutDownCluster(); |
| startUpCluster(); |
| |
| try { |
| ArrayList<Block> blocks = |
| writeFile(METHOD_NAME, 12 * bytesChkSum, filePath); |
| Block bl = findBlock(filePath, 12 * bytesChkSum); |
| BlockChecker bc = new BlockChecker(filePath); |
| bc.start(); |
| |
| waitForTempReplica(bl, DN_N1); |
| |
| // all blocks belong to the same file, hence same BP |
| DataNode dn = cluster.getDataNodes().get(DN_N1); |
| String poolId = cluster.getNamesystem().getBlockPoolId(); |
| DatanodeRegistration dnR = dn.getDNRegistrationForBP(poolId); |
| StorageBlockReport[] report = { new StorageBlockReport(dnR.getStorageID(), |
| new BlockListAsLongs(blocks, null).getBlockListAsLongs()) }; |
| cluster.getNameNodeRpc().blockReport(dnR, poolId, report); |
| printStats(); |
| assertEquals("Wrong number of PendingReplication blocks", |
| blocks.size(), cluster.getNamesystem().getPendingReplicationBlocks()); |
| |
| try { |
| bc.join(); |
| } catch (InterruptedException e) { } |
| } finally { |
| resetConfiguration(); // return the initial state of the configuration |
| } |
| } |
| |
| // Similar to BlockReport_08 but corrupts GS and len of the TEMPORARY's |
| // replica block. Expect the same behaviour: NN should simply ignore this |
| // block |
| @Test |
| public void blockReport_09() throws IOException { |
| final String METHOD_NAME = GenericTestUtils.getMethodName(); |
| Path filePath = new Path("/" + METHOD_NAME + ".dat"); |
| final int DN_N1 = DN_N0 + 1; |
| final int bytesChkSum = 1024 * 1000; |
| |
| conf.setInt(DFSConfigKeys.DFS_BYTES_PER_CHECKSUM_KEY, bytesChkSum); |
| conf.setLong(DFSConfigKeys.DFS_BLOCK_SIZE_KEY, 6 * bytesChkSum); |
| shutDownCluster(); |
| startUpCluster(); |
| // write file and start second node to be "older" than the original |
| |
| try { |
| ArrayList<Block> blocks = |
| writeFile(METHOD_NAME, 12 * bytesChkSum, filePath); |
| |
| Block bl = findBlock(filePath, 12 * bytesChkSum); |
| BlockChecker bc = new BlockChecker(filePath); |
| bc.start(); |
| corruptBlockGS(bl); |
| corruptBlockLen(bl); |
| |
| waitForTempReplica(bl, DN_N1); |
| |
| // all blocks belong to the same file, hence same BP |
| DataNode dn = cluster.getDataNodes().get(DN_N1); |
| String poolId = cluster.getNamesystem().getBlockPoolId(); |
| DatanodeRegistration dnR = dn.getDNRegistrationForBP(poolId); |
| StorageBlockReport[] report = { new StorageBlockReport(dnR.getStorageID(), |
| new BlockListAsLongs(blocks, null).getBlockListAsLongs()) }; |
| cluster.getNameNodeRpc().blockReport(dnR, poolId, report); |
| printStats(); |
| assertEquals("Wrong number of PendingReplication blocks", |
| 2, cluster.getNamesystem().getPendingReplicationBlocks()); |
| |
| try { |
| bc.join(); |
| } catch (InterruptedException e) {} |
| } finally { |
| resetConfiguration(); // return the initial state of the configuration |
| } |
| } |
| |
| /** |
| * Test for the case where one of the DNs in the pipeline is in the |
| * process of doing a block report exactly when the block is closed. |
| * In this case, the block report becomes delayed until after the |
| * block is marked completed on the NN, and hence it reports an RBW |
| * replica for a COMPLETE block. Such a report should not be marked |
| * corrupt. |
| * This is a regression test for HDFS-2791. |
| */ |
| @Test |
| public void testOneReplicaRbwReportArrivesAfterBlockCompleted() throws Exception { |
| final CountDownLatch brFinished = new CountDownLatch(1); |
| DelayAnswer delayer = new GenericTestUtils.DelayAnswer(LOG) { |
| @Override |
| protected Object passThrough(InvocationOnMock invocation) |
| throws Throwable { |
| try { |
| return super.passThrough(invocation); |
| } finally { |
| // inform the test that our block report went through. |
| brFinished.countDown(); |
| } |
| } |
| }; |
| |
| final String METHOD_NAME = GenericTestUtils.getMethodName(); |
| Path filePath = new Path("/" + METHOD_NAME + ".dat"); |
| |
| // Start a second DN for this test -- we're checking |
| // what happens when one of the DNs is slowed for some reason. |
| REPL_FACTOR = 2; |
| startDNandWait(null, false); |
| |
| NameNode nn = cluster.getNameNode(); |
| |
| FSDataOutputStream out = fs.create(filePath, REPL_FACTOR); |
| try { |
| AppendTestUtil.write(out, 0, 10); |
| out.hflush(); |
| |
| // Set up a spy so that we can delay the block report coming |
| // from this node. |
| DataNode dn = cluster.getDataNodes().get(0); |
| DatanodeProtocol spy = |
| DataNodeAdapter.spyOnBposToNN(dn, nn); |
| |
| Mockito.doAnswer(delayer) |
| .when(spy).blockReport( |
| Mockito.<DatanodeRegistration>anyObject(), |
| Mockito.anyString(), |
| Mockito.<StorageBlockReport[]>anyObject()); |
| |
| // Force a block report to be generated. The block report will have |
| // an RBW replica in it. Wait for the RPC to be sent, but block |
| // it before it gets to the NN. |
| dn.scheduleAllBlockReport(0); |
| delayer.waitForCall(); |
| |
| } finally { |
| IOUtils.closeStream(out); |
| } |
| |
| // Now that the stream is closed, the NN will have the block in COMPLETE |
| // state. |
| delayer.proceed(); |
| brFinished.await(); |
| |
| // Verify that no replicas are marked corrupt, and that the |
| // file is still readable. |
| BlockManagerTestUtil.updateState(nn.getNamesystem().getBlockManager()); |
| assertEquals(0, nn.getNamesystem().getCorruptReplicaBlocks()); |
| DFSTestUtil.readFile(fs, filePath); |
| |
| // Ensure that the file is readable even from the DN that we futzed with. |
| cluster.stopDataNode(1); |
| DFSTestUtil.readFile(fs, filePath); |
| } |
| |
| private void waitForTempReplica(Block bl, int DN_N1) throws IOException { |
| final boolean tooLongWait = false; |
| final int TIMEOUT = 40000; |
| |
| if(LOG.isDebugEnabled()) { |
| LOG.debug("Wait for datanode " + DN_N1 + " to appear"); |
| } |
| while (cluster.getDataNodes().size() <= DN_N1) { |
| waitTil(20); |
| } |
| if(LOG.isDebugEnabled()) { |
| LOG.debug("Total number of DNs " + cluster.getDataNodes().size()); |
| } |
| cluster.waitActive(); |
| |
| // Look about specified DN for the replica of the block from 1st DN |
| String bpid = cluster.getNamesystem().getBlockPoolId(); |
| Replica r = ((FSDataset) cluster.getDataNodes().get(DN_N1).getFSDataset()). |
| fetchReplicaInfo(bpid, bl.getBlockId()); |
| long start = System.currentTimeMillis(); |
| int count = 0; |
| while (r == null) { |
| waitTil(5); |
| r = ((FSDataset) cluster.getDataNodes().get(DN_N1).getFSDataset()). |
| fetchReplicaInfo(bpid, bl.getBlockId()); |
| long waiting_period = System.currentTimeMillis() - start; |
| if (count++ % 100 == 0) |
| if(LOG.isDebugEnabled()) { |
| LOG.debug("Has been waiting for " + waiting_period + " ms."); |
| } |
| if (waiting_period > TIMEOUT) |
| assertTrue("Was waiting too long to get ReplicaInfo from a datanode", |
| tooLongWait); |
| } |
| |
| HdfsServerConstants.ReplicaState state = r.getState(); |
| if(LOG.isDebugEnabled()) { |
| LOG.debug("Replica state before the loop " + state.getValue()); |
| } |
| start = System.currentTimeMillis(); |
| while (state != HdfsServerConstants.ReplicaState.TEMPORARY) { |
| waitTil(5); |
| state = r.getState(); |
| if(LOG.isDebugEnabled()) { |
| LOG.debug("Keep waiting for " + bl.getBlockName() + |
| " is in state " + state.getValue()); |
| } |
| if (System.currentTimeMillis() - start > TIMEOUT) |
| assertTrue("Was waiting too long for a replica to become TEMPORARY", |
| tooLongWait); |
| } |
| if(LOG.isDebugEnabled()) { |
| LOG.debug("Replica state after the loop " + state.getValue()); |
| } |
| } |
| |
| // Helper methods from here below... |
| // Write file and start second data node. |
| private ArrayList<Block> writeFile(final String METHOD_NAME, |
| final long fileSize, |
| Path filePath) { |
| ArrayList<Block> blocks = null; |
| try { |
| REPL_FACTOR = 2; |
| blocks = prepareForRide(filePath, METHOD_NAME, fileSize); |
| } catch (IOException e) { |
| if(LOG.isDebugEnabled()) { |
| LOG.debug("Caught exception ", e); |
| } |
| } |
| return blocks; |
| } |
| |
| private void startDNandWait(Path filePath, boolean waitReplicas) |
| throws IOException { |
| if(LOG.isDebugEnabled()) { |
| LOG.debug("Before next DN start: " + cluster.getDataNodes().size()); |
| } |
| cluster.startDataNodes(conf, 1, true, null, null); |
| ArrayList<DataNode> datanodes = cluster.getDataNodes(); |
| assertEquals(datanodes.size(), 2); |
| |
| if(LOG.isDebugEnabled()) { |
| LOG.debug("New datanode " |
| + cluster.getDataNodes().get(datanodes.size() - 1).getMachineName() |
| + " has been started"); |
| } |
| if (waitReplicas) DFSTestUtil.waitReplication(fs, filePath, REPL_FACTOR); |
| } |
| |
| private ArrayList<Block> prepareForRide(final Path filePath, |
| final String METHOD_NAME, |
| long fileSize) throws IOException { |
| LOG.info("Running test " + METHOD_NAME); |
| |
| DFSTestUtil.createFile(fs, filePath, fileSize, |
| REPL_FACTOR, rand.nextLong()); |
| |
| return locatedToBlocks(cluster.getNameNodeRpc() |
| .getBlockLocations(filePath.toString(), FILE_START, |
| fileSize).getLocatedBlocks(), null); |
| } |
| |
| private void printStats() { |
| BlockManagerTestUtil.updateState(cluster.getNamesystem().getBlockManager()); |
| if(LOG.isDebugEnabled()) { |
| LOG.debug("Missing " + cluster.getNamesystem().getMissingBlocksCount()); |
| LOG.debug("Corrupted " + cluster.getNamesystem().getCorruptReplicaBlocks()); |
| LOG.debug("Under-replicated " + cluster.getNamesystem(). |
| getUnderReplicatedBlocks()); |
| LOG.debug("Pending delete " + cluster.getNamesystem(). |
| getPendingDeletionBlocks()); |
| LOG.debug("Pending replications " + cluster.getNamesystem(). |
| getPendingReplicationBlocks()); |
| LOG.debug("Excess " + cluster.getNamesystem().getExcessBlocks()); |
| LOG.debug("Total " + cluster.getNamesystem().getBlocksTotal()); |
| } |
| } |
| |
| private ArrayList<Block> locatedToBlocks(final List<LocatedBlock> locatedBlks, |
| List<Integer> positionsToRemove) { |
| ArrayList<Block> newList = new ArrayList<Block>(); |
| for (int i = 0; i < locatedBlks.size(); i++) { |
| if (positionsToRemove != null && positionsToRemove.contains(i)) { |
| if(LOG.isDebugEnabled()) { |
| LOG.debug(i + " block to be omitted"); |
| } |
| continue; |
| } |
| newList.add(new Block(locatedBlks.get(i).getBlock().getLocalBlock())); |
| } |
| return newList; |
| } |
| |
| private void waitTil(long waitPeriod) { |
| try { //Wait til next re-scan |
| Thread.sleep(waitPeriod); |
| } catch (InterruptedException e) { |
| e.printStackTrace(); |
| } |
| } |
| |
| private List<File> findAllFiles(File top, FilenameFilter mask) { |
| if (top == null) return null; |
| ArrayList<File> ret = new ArrayList<File>(); |
| for (File f : top.listFiles()) { |
| if (f.isDirectory()) |
| ret.addAll(findAllFiles(f, mask)); |
| else if (mask.accept(f, f.getName())) |
| ret.add(f); |
| } |
| return ret; |
| } |
| |
| private class MyFileFilter implements FilenameFilter { |
| private String nameToAccept = ""; |
| private boolean all = false; |
| |
| public MyFileFilter(String nameToAccept, boolean all) { |
| if (nameToAccept == null) |
| throw new IllegalArgumentException("Argument isn't suppose to be null"); |
| this.nameToAccept = nameToAccept; |
| this.all = all; |
| } |
| |
| public boolean accept(File file, String s) { |
| if (all) |
| return s != null && s.startsWith(nameToAccept); |
| else |
| return s != null && s.equals(nameToAccept); |
| } |
| } |
| |
| private static void initLoggers() { |
| ((Log4JLogger) NameNode.stateChangeLog).getLogger().setLevel(Level.ALL); |
| ((Log4JLogger) LogFactory.getLog(FSNamesystem.class)).getLogger().setLevel(Level.ALL); |
| ((Log4JLogger) DataNode.LOG).getLogger().setLevel(Level.ALL); |
| ((Log4JLogger) TestBlockReport.LOG).getLogger().setLevel(Level.ALL); |
| } |
| |
| private void corruptBlockLen(final Block block) |
| throws IOException { |
| if (block == null) { |
| throw new IOException("Block isn't suppose to be null"); |
| } |
| long oldLen = block.getNumBytes(); |
| long newLen = oldLen - rand.nextLong(); |
| assertTrue("Old and new length shouldn't be the same", |
| block.getNumBytes() != newLen); |
| block.setNumBytes(newLen); |
| if(LOG.isDebugEnabled()) { |
| LOG.debug("Length of " + block.getBlockName() + |
| " is changed to " + newLen + " from " + oldLen); |
| } |
| } |
| |
| private void corruptBlockGS(final Block block) |
| throws IOException { |
| if (block == null) { |
| throw new IOException("Block isn't suppose to be null"); |
| } |
| long oldGS = block.getGenerationStamp(); |
| long newGS = oldGS - rand.nextLong(); |
| assertTrue("Old and new GS shouldn't be the same", |
| block.getGenerationStamp() != newGS); |
| block.setGenerationStamp(newGS); |
| if(LOG.isDebugEnabled()) { |
| LOG.debug("Generation stamp of " + block.getBlockName() + |
| " is changed to " + block.getGenerationStamp() + " from " + oldGS); |
| } |
| } |
| |
| private Block findBlock(Path path, long size) throws IOException { |
| Block ret; |
| List<LocatedBlock> lbs = |
| cluster.getNameNodeRpc() |
| .getBlockLocations(path.toString(), |
| FILE_START, size).getLocatedBlocks(); |
| LocatedBlock lb = lbs.get(lbs.size() - 1); |
| |
| // Get block from the first DN |
| ret = cluster.getDataNodes().get(DN_N0). |
| data.getStoredBlock(lb.getBlock() |
| .getBlockPoolId(), lb.getBlock().getBlockId()); |
| return ret; |
| } |
| |
| private class BlockChecker extends Thread { |
| Path filePath; |
| |
| public BlockChecker(final Path filePath) { |
| this.filePath = filePath; |
| } |
| |
| public void run() { |
| try { |
| startDNandWait(filePath, true); |
| } catch (IOException e) { |
| LOG.warn("Shouldn't happen", e); |
| } |
| } |
| } |
| |
| private static void resetConfiguration() { |
| conf = new Configuration(); |
| int customPerChecksumSize = 512; |
| int customBlockSize = customPerChecksumSize * 3; |
| conf.setInt(DFSConfigKeys.DFS_BYTES_PER_CHECKSUM_KEY, customPerChecksumSize); |
| conf.setLong(DFSConfigKeys.DFS_BLOCK_SIZE_KEY, customBlockSize); |
| conf.setLong(DFSConfigKeys.DFS_DATANODE_DIRECTORYSCAN_INTERVAL_KEY, DN_RESCAN_INTERVAL); |
| } |
| } |
