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);
  }
@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);
    // Back up the uncorrupted log
    File logFileBak = new File(testDir, logFile.getName() + ".bak");
    Files.copy(logFile, logFileBak);
    EditLogValidation validation =
        EditLogFileInputStream.scanEditLog(logFile, Long.MAX_VALUE, true);
    assertTrue(!validation.hasCorruptHeader());
    // We expect that there will be an OP_START_LOG_SEGMENT, followed by
    // NUM_TXNS opcodes, followed by an OP_END_LOG_SEGMENT.
    assertEquals(NUM_TXNS + 1, validation.getEndTxId());
    // Corrupt each edit and verify that validation continues to work
    for (Map.Entry<Long, Long> entry : offsetToTxId.entrySet()) {
      long txOffset = entry.getKey();
      long txId = entry.getValue();

      // Restore backup, corrupt the txn opcode
      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());
    }

    // Truncate right before each edit and verify that validation continues
    // to work
    for (Map.Entry<Long, Long> entry : offsetToTxId.entrySet()) {
      long txOffset = entry.getKey();
      long txId = entry.getValue();

      // Restore backup, corrupt the txn opcode
      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());
    }
  }