EditLogTailer { @VisibleForTesting FSEditLog getEditLog() { return editLog; } EditLogTailer(FSNamesystem namesystem, Configuration conf); void start(); void stop(); @VisibleForTesting void setEditLog(FSEditLog editLog); void catchupDuringFailover(); @VisibleForTesting long doTailEdits(); long getLastLoadTimeMs(); static final Logger LOG; static final String DFS_HA_TAILEDITS_MAX_TXNS_PER_LOCK_KEY; static final long DFS_HA_TAILEDITS_MAX_TXNS_PER_LOCK_DEFAULT; }
@Test public void testTailer() throws IOException, InterruptedException, ServiceFailedException { Configuration conf = getConf(); conf.setInt(DFSConfigKeys.DFS_HA_TAILEDITS_PERIOD_KEY, 0); conf.setInt(DFSConfigKeys.DFS_HA_TAILEDITS_ALL_NAMESNODES_RETRY_KEY, 100); conf.setLong(EditLogTailer.DFS_HA_TAILEDITS_MAX_TXNS_PER_LOCK_KEY, 3); HAUtil.setAllowStandbyReads(conf, true); MiniDFSCluster cluster = new MiniDFSCluster.Builder(conf) .nnTopology(MiniDFSNNTopology.simpleHATopology()) .numDataNodes(0) .build(); cluster.waitActive(); cluster.transitionToActive(0); NameNode nn1 = cluster.getNameNode(0); NameNode nn2 = cluster.getNameNode(1); try { for (int i = 0; i < DIRS_TO_MAKE / 2; i++) { NameNodeAdapter.mkdirs(nn1, getDirPath(i), new PermissionStatus("test","test", new FsPermission((short)00755)), true); } HATestUtil.waitForStandbyToCatchUp(nn1, nn2); assertEquals("Inconsistent number of applied txns on Standby", nn1.getNamesystem().getEditLog().getLastWrittenTxId(), nn2.getNamesystem().getFSImage().getLastAppliedTxId() + 1); for (int i = 0; i < DIRS_TO_MAKE / 2; i++) { assertTrue(NameNodeAdapter.getFileInfo(nn2, getDirPath(i), false, false, false).isDirectory()); } for (int i = DIRS_TO_MAKE / 2; i < DIRS_TO_MAKE; i++) { NameNodeAdapter.mkdirs(nn1, getDirPath(i), new PermissionStatus("test","test", new FsPermission((short)00755)), true); } HATestUtil.waitForStandbyToCatchUp(nn1, nn2); assertEquals("Inconsistent number of applied txns on Standby", nn1.getNamesystem().getEditLog().getLastWrittenTxId(), nn2.getNamesystem().getFSImage().getLastAppliedTxId() + 1); for (int i = DIRS_TO_MAKE / 2; i < DIRS_TO_MAKE; i++) { assertTrue(NameNodeAdapter.getFileInfo(nn2, getDirPath(i), false, false, false).isDirectory()); } } finally { cluster.shutdown(); } }
@Test public void testStandbyTriggersLogRollsWhenTailInProgressEdits() throws Exception { final int standbyCatchupWaitTime = 2; final int noLogRollWaitTime = 2; final int logRollWaitTime = 3; Configuration conf = getConf(); conf.setInt(DFSConfigKeys.DFS_HA_LOGROLL_PERIOD_KEY, standbyCatchupWaitTime + noLogRollWaitTime + 1); conf.setInt(DFSConfigKeys.DFS_HA_TAILEDITS_PERIOD_KEY, 1); conf.setBoolean(DFSConfigKeys.DFS_HA_TAILEDITS_INPROGRESS_KEY, true); MiniDFSCluster cluster = createMiniDFSCluster(conf, 2); if (cluster == null) { fail("failed to start mini cluster."); } try { int activeIndex = new Random().nextBoolean() ? 1 : 0; int standbyIndex = (activeIndex == 0) ? 1 : 0; cluster.transitionToActive(activeIndex); NameNode active = cluster.getNameNode(activeIndex); NameNode standby = cluster.getNameNode(standbyIndex); long origTxId = active.getNamesystem().getFSImage().getEditLog() .getCurSegmentTxId(); for (int i = 0; i < DIRS_TO_MAKE / 2; i++) { NameNodeAdapter.mkdirs(active, getDirPath(i), new PermissionStatus("test", "test", new FsPermission((short)00755)), true); } long activeTxId = active.getNamesystem().getFSImage().getEditLog() .getLastWrittenTxId(); waitForStandbyToCatchUpWithInProgressEdits(standby, activeTxId, standbyCatchupWaitTime); for (int i = DIRS_TO_MAKE / 2; i < DIRS_TO_MAKE; i++) { NameNodeAdapter.mkdirs(active, getDirPath(i), new PermissionStatus("test", "test", new FsPermission((short)00755)), true); } boolean exceptionThrown = false; try { checkForLogRoll(active, origTxId, noLogRollWaitTime); } catch (TimeoutException e) { exceptionThrown = true; } assertTrue(exceptionThrown); checkForLogRoll(active, origTxId, logRollWaitTime); } finally { cluster.shutdown(); } }