FSEditLogLoader { static EditLogValidation scanEditLog(EditLogInputStream in, long maxTxIdToScan) { long lastPos; long lastTxId = HdfsServerConstants.INVALID_TXID; long numValid = 0; while (true) { long txid; lastPos = in.getPosition(); try { if ((txid = in.scanNextOp()) == HdfsServerConstants.INVALID_TXID) { break; } } catch (Throwable t) { FSImage.LOG.warn("Caught exception after scanning through " + numValid + " ops from " + in + " while determining its valid length. Position was " + lastPos, t); in.resync(); FSImage.LOG.warn("After resync, position is " + in.getPosition()); if (in.getPosition() <= lastPos) { FSImage.LOG.warn("After resync, the position, {} is not greater " + "than the previous position {}. Skipping remainder of this log.", in.getPosition(), lastPos); break; } continue; } if (lastTxId == HdfsServerConstants.INVALID_TXID || txid > lastTxId) { lastTxId = txid; } if (lastTxId >= maxTxIdToScan) { break; } numValid++; } return new EditLogValidation(lastPos, lastTxId, false); } FSEditLogLoader(FSNamesystem fsNamesys, long lastAppliedTxId); @VisibleForTesting  FSEditLogLoader(FSNamesystem fsNamesys, long lastAppliedTxId, Timer timer); long getLastAppliedTxId();  }
@Test public void testValidateEditLogWithCorruptHeader() throws IOException { File testDir = new File(TEST_DIR, "testValidateEditLogWithCorruptHeader"); SortedMap<Long, Long> offsetToTxId = Maps.newTreeMap(); File logFile = prepareUnfinalizedTestEditLog(testDir, 2, offsetToTxId); RandomAccessFile rwf = new RandomAccessFile(logFile, "rw"); try { rwf.seek(0); rwf.writeLong(42); } finally { rwf.close(); } EditLogValidation validation = EditLogFileInputStream.scanEditLog(logFile, Long.MAX_VALUE, true); assertTrue(validation.hasCorruptHeader()); }
@Test public void testValidateEditLogWithCorruptBody() throws IOException { File testDir = new File(TEST_DIR, "testValidateEditLogWithCorruptBody"); SortedMap<Long, Long> offsetToTxId = Maps.newTreeMap(); final int NUM_TXNS = 20; File logFile = prepareUnfinalizedTestEditLog(testDir, NUM_TXNS, offsetToTxId); File logFileBak = new File(testDir, logFile.getName() + ".bak"); Files.copy(logFile, logFileBak); EditLogValidation validation = EditLogFileInputStream.scanEditLog(logFile, Long.MAX_VALUE, true); assertTrue(!validation.hasCorruptHeader()); assertEquals(NUM_TXNS + 1, validation.getEndTxId()); for (Map.Entry<Long, Long> entry : offsetToTxId.entrySet()) { long txOffset = entry.getKey(); long txId = entry.getValue(); Files.copy(logFileBak, logFile); corruptByteInFile(logFile, txOffset); validation = EditLogFileInputStream.scanEditLog(logFile, Long.MAX_VALUE, true); long expectedEndTxId = (txId == (NUM_TXNS + 1)) ? NUM_TXNS : (NUM_TXNS + 1); assertEquals("Failed when corrupting txn opcode at " + txOffset, expectedEndTxId, validation.getEndTxId()); assertTrue(!validation.hasCorruptHeader()); } for (Map.Entry<Long, Long> entry : offsetToTxId.entrySet()) { long txOffset = entry.getKey(); long txId = entry.getValue(); Files.copy(logFileBak, logFile); truncateFile(logFile, txOffset); validation = EditLogFileInputStream.scanEditLog(logFile, Long.MAX_VALUE, true); long expectedEndTxId = (txId == 0) ? HdfsServerConstants.INVALID_TXID : (txId - 1); assertEquals("Failed when corrupting txid " + txId + " txn opcode " + "at " + txOffset, expectedEndTxId, validation.getEndTxId()); assertTrue(!validation.hasCorruptHeader()); } }
@Test public void testValidateEmptyEditLog() throws IOException { File testDir = new File(TEST_DIR, "testValidateEmptyEditLog"); SortedMap<Long, Long> offsetToTxId = Maps.newTreeMap(); File logFile = prepareUnfinalizedTestEditLog(testDir, 0, offsetToTxId); truncateFile(logFile, 8); EditLogValidation validation = EditLogFileInputStream.scanEditLog(logFile, Long.MAX_VALUE, true); assertTrue(!validation.hasCorruptHeader()); assertEquals(HdfsServerConstants.INVALID_TXID, validation.getEndTxId()); }