@Test(timeout=120000) public void testMarkSuspectBlock() throws Exception { Configuration conf = new Configuration(); conf.setLong(DFS_DATANODE_SCAN_PERIOD_HOURS_KEY, 100L); conf.set(INTERNAL_VOLUME_SCANNER_SCAN_RESULT_HANDLER, TestScanResultHandler.class.getName()); conf.setLong(INTERNAL_DFS_BLOCK_SCANNER_CURSOR_SAVE_INTERVAL_MS, 0L); final TestContext ctx = new TestContext(conf, 1); final int NUM_EXPECTED_BLOCKS = 10; ctx.createFiles(0, NUM_EXPECTED_BLOCKS, 1); final TestScanResultHandler.Info info = TestScanResultHandler.getInfo(ctx.volumes.get(0)); String storageID = ctx.datanode.getFSDataset(). getVolumes().get(0).getStorageID(); synchronized (info) { info.sem = new Semaphore(4); info.shouldRun = true; info.notify(); } LOG.info("Waiting for the first 4 blocks to be scanned."); GenericTestUtils.waitFor(new Supplier<Boolean>() { @Override public Boolean get() { synchronized (info) { if (info.blocksScanned >= 4) { LOG.info("info = {}. blockScanned has now reached 4.", info); return true; } else { LOG.info("info = {}. Waiting for blockScanned to reach 4.", info); return false; } } } }, 50, 30000); synchronized (info) { assertEquals("Expected 4 good blocks.", 4, info.goodBlocks.size()); info.goodBlocks.clear(); assertEquals("Expected 4 blocksScanned", 4, info.blocksScanned); assertEquals("Did not expect bad blocks.", 0, info.badBlocks.size()); info.blocksScanned = 0; } ExtendedBlock first = ctx.getFileBlock(0, 0); ctx.datanode.getBlockScanner().markSuspectBlock(storageID, first); info.sem.release(2); LOG.info("Waiting for 2 more blocks to be scanned."); GenericTestUtils.waitFor(new Supplier<Boolean>() { @Override public Boolean get() { synchronized (info) { if (info.blocksScanned >= 2) { LOG.info("info = {}. blockScanned has now reached 2.", info); return true; } else { LOG.info("info = {}. Waiting for blockScanned to reach 2.", info); return false; } } } }, 50, 30000); synchronized (info) { assertTrue("Expected block " + first + " to have been scanned.", info.goodBlocks.contains(first)); assertEquals(2, info.goodBlocks.size()); info.goodBlocks.clear(); assertEquals("Did not expect bad blocks.", 0, info.badBlocks.size()); assertEquals(2, info.blocksScanned); info.blocksScanned = 0; } ctx.datanode.getBlockScanner().markSuspectBlock(storageID, first); info.sem.release(10); LOG.info("Waiting for 5 more blocks to be scanned."); GenericTestUtils.waitFor(new Supplier<Boolean>() { @Override public Boolean get() { synchronized (info) { if (info.blocksScanned >= 5) { LOG.info("info = {}. blockScanned has now reached 5.", info); return true; } else { LOG.info("info = {}. Waiting for blockScanned to reach 5.", info); return false; } } } }, 50, 30000); synchronized (info) { assertEquals(5, info.goodBlocks.size()); assertEquals(0, info.badBlocks.size()); assertEquals(5, info.blocksScanned); Assert.assertFalse("We should not " + "have rescanned block " + first + ", because it should have been " + "in recentSuspectBlocks.", info.goodBlocks.contains(first)); info.blocksScanned = 0; } }
synchronized void markSuspectBlock(String storageId, ExtendedBlock block) { if (!isEnabled()) { LOG.debug("Not scanning suspicious block {} on {}, because the block " + "scanner is disabled.", block, storageId); return; } VolumeScanner scanner = scanners.get(storageId); if (scanner == null) { LOG.info("Not scanning suspicious block {} on {}, because there is no " + "volume scanner for that storageId.", block, storageId); return; } scanner.markSuspectBlock(block); }
BlockScanner { synchronized void markSuspectBlock(String storageId, ExtendedBlock block) { if (!isEnabled()) { LOG.debug("Not scanning suspicious block {} on {}, because the block " + "scanner is disabled.", block, storageId); return; } VolumeScanner scanner = scanners.get(storageId); if (scanner == null) { LOG.info("Not scanning suspicious block {} on {}, because there is no " + "volume scanner for that storageId.", block, storageId); return; } scanner.markSuspectBlock(block); } }
BlockScanner { synchronized void markSuspectBlock(String storageId, ExtendedBlock block) { if (!isEnabled()) { LOG.debug("Not scanning suspicious block {} on {}, because the block " + "scanner is disabled.", block, storageId); return; } VolumeScanner scanner = scanners.get(storageId); if (scanner == null) { LOG.info("Not scanning suspicious block {} on {}, because there is no " + "volume scanner for that storageId.", block, storageId); return; } scanner.markSuspectBlock(block); } BlockScanner(DataNode datanode, Configuration conf); }
BlockScanner { synchronized void markSuspectBlock(String storageId, ExtendedBlock block) { if (!isEnabled()) { LOG.debug("Not scanning suspicious block {} on {}, because the block " + "scanner is disabled.", block, storageId); return; } VolumeScanner scanner = scanners.get(storageId); if (scanner == null) { LOG.info("Not scanning suspicious block {} on {}, because there is no " + "volume scanner for that storageId.", block, storageId); return; } scanner.markSuspectBlock(block); } BlockScanner(DataNode datanode, Configuration conf); boolean isEnabled(); synchronized void addVolumeScanner(FsVolumeReference ref); synchronized void removeVolumeScanner(FsVolumeSpi volume); synchronized void removeAllVolumeScanners(); }
BlockScanner { synchronized void markSuspectBlock(String storageId, ExtendedBlock block) { if (!isEnabled()) { LOG.debug("Not scanning suspicious block {} on {}, because the block " + "scanner is disabled.", block, storageId); return; } VolumeScanner scanner = scanners.get(storageId); if (scanner == null) { LOG.info("Not scanning suspicious block {} on {}, because there is no " + "volume scanner for that storageId.", block, storageId); return; } scanner.markSuspectBlock(block); } BlockScanner(DataNode datanode, Configuration conf); boolean isEnabled(); synchronized void addVolumeScanner(FsVolumeReference ref); synchronized void removeVolumeScanner(FsVolumeSpi volume); synchronized void removeAllVolumeScanners(); static final Logger LOG; }