HDFS-8996. Consolidate validateLog and scanLog in FJM#EditLogFile (Zhe Zhang via Colin P. McCabe)

(cherry picked from commit 53bad4eb00)

hadoop-hdfs-project/hadoop-hdfs/CHANGES.txt

@@ -558,6 +558,9 @@ Release 2.8.0 - UNRELEASED
     HDFS-9027. Refactor o.a.h.hdfs.DataStreamer#isLazyPersist() method.
     (Mingliang Liu via Arpit Agarwal)
 
+    HDFS-8996. Consolidate validateLog and scanLog in FJM#EditLogFile (Zhe
+    Zhang via Colin P. McCabe)
+
   OPTIMIZATIONS
 
     HDFS-8026. Trace FSOutputSummer#writeChecksumChunks rather than

hadoop-hdfs-project/hadoop-hdfs/src/contrib/bkjournal/src/test/java/org/apache/hadoop/hdfs/server/namenode/FSEditLogTestUtil.java

@@ -34,7 +34,7 @@ public class FSEditLogTestUtil {
   public static long countTransactionsInStream(EditLogInputStream in) 
       throws IOException {
     FSEditLogLoader.EditLogValidation validation =
-        FSEditLogLoader.validateEditLog(in, Long.MAX_VALUE);
+        FSEditLogLoader.scanEditLog(in, Long.MAX_VALUE);
     return (validation.getEndTxId() - in.getFirstTxId()) + 1;
   }
 }

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/qjournal/server/Journal.java

@@ -192,7 +192,7 @@ public class Journal implements Closeable {
     
     while (!files.isEmpty()) {
       EditLogFile latestLog = files.remove(files.size() - 1);
-      latestLog.scanLog();
+      latestLog.scanLog(Long.MAX_VALUE, false);
       LOG.info("Latest log is " + latestLog);
       if (latestLog.getLastTxId() == HdfsServerConstants.INVALID_TXID) {
         // the log contains no transactions
@@ -542,7 +542,7 @@ public class Journal implements Closeable {
       // If it's in-progress, it should only contain one transaction,
       // because the "startLogSegment" transaction is written alone at the
       // start of each segment. 
-      existing.scanLog();
+      existing.scanLog(Long.MAX_VALUE, false);
       if (existing.getLastTxId() != existing.getFirstTxId()) {
         throw new IllegalStateException("The log file " +
             existing + " seems to contain valid transactions");
@@ -605,7 +605,7 @@ public class Journal implements Closeable {
       if (needsValidation) {
         LOG.info("Validating log segment " + elf.getFile() + " about to be " +
             "finalized");
-        elf.scanLog();
+        elf.scanLog(Long.MAX_VALUE, false);
   
         checkSync(elf.getLastTxId() == endTxId,
             "Trying to finalize in-progress log segment %s to end at " +
@@ -693,7 +693,7 @@ public class Journal implements Closeable {
       return null;
     }
     if (elf.isInProgress()) {
-      elf.scanLog();
+      elf.scanLog(Long.MAX_VALUE, false);
     }
     if (elf.getLastTxId() == HdfsServerConstants.INVALID_TXID) {
       LOG.info("Edit log file " + elf + " appears to be empty. " +

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/EditLogFileInputStream.java

@@ -302,74 +302,31 @@ public class EditLogFileInputStream extends EditLogInputStream {
   }
 
   /**
-   * @param file File being validated.
+   * @param file          File being scanned and validated.
-   * @param maxTxIdToValidate Maximum Tx ID to try to validate. Validation
+   * @param maxTxIdToScan Maximum Tx ID to try to scan.
-   *                          returns after reading this or a higher ID.
+   *                      The scan returns after reading this or a higher
-   *                          The file portion beyond this ID is potentially
+   *                      ID. The file portion beyond this ID is
-   *                          being updated.
+   *                      potentially being updated.
    * @return Result of the validation
    * @throws IOException
    */
-  static FSEditLogLoader.EditLogValidation validateEditLog(File file,
+  static FSEditLogLoader.EditLogValidation scanEditLog(File file,
-      long maxTxIdToValidate) throws IOException {
+      long maxTxIdToScan, boolean verifyVersion)
-    EditLogFileInputStream in;
-    try {
-      in = new EditLogFileInputStream(file);
-      in.getVersion(true); // causes us to read the header
-    } catch (LogHeaderCorruptException e) {
-      // If the header is malformed or the wrong value, this indicates a corruption
-      LOG.warn("Log file " + file + " has no valid header", e);
-      return new FSEditLogLoader.EditLogValidation(0,
-          HdfsServerConstants.INVALID_TXID, true);
-    }
-    
-    try {
-      return FSEditLogLoader.validateEditLog(in, maxTxIdToValidate);
-    } finally {
-      IOUtils.closeStream(in);
-    }
-  }
-
-  static FSEditLogLoader.EditLogValidation scanEditLog(File file)
       throws IOException {
     EditLogFileInputStream in;
     try {
       in = new EditLogFileInputStream(file);
       // read the header, initialize the inputstream, but do not check the
       // layoutversion
-      in.getVersion(false);
+      in.getVersion(verifyVersion);
     } catch (LogHeaderCorruptException e) {
       LOG.warn("Log file " + file + " has no valid header", e);
       return new FSEditLogLoader.EditLogValidation(0,
           HdfsServerConstants.INVALID_TXID, true);
     }
 
-    long lastPos = 0;
-    long lastTxId = HdfsServerConstants.INVALID_TXID;
-    long numValid = 0;
     try {
-      while (true) {
+      return FSEditLogLoader.scanEditLog(in, maxTxIdToScan);
-        long txid = HdfsServerConstants.INVALID_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());
-          continue;
-        }
-        if (lastTxId == HdfsServerConstants.INVALID_TXID || txid > lastTxId) {
-          lastTxId = txid;
-        }
-        numValid++;
-      }
-      return new EditLogValidation(lastPos, lastTxId, false);
     } finally {
       IOUtils.closeStream(in);
     }

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSEditLogLoader.java

@@ -1113,70 +1113,41 @@ public class FSEditLogLoader {
   /**
    * Find the last valid transaction ID in the stream.
    * If there are invalid or corrupt transactions in the middle of the stream,
-   * validateEditLog will skip over them.
+   * scanEditLog will skip over them.
    * This reads through the stream but does not close it.
    *
-   * @param maxTxIdToValidate Maximum Tx ID to try to validate. Validation
+   * @param maxTxIdToScan Maximum Tx ID to try to scan.
-   *                          returns after reading this or a higher ID.
+   *                      The scan returns after reading this or a higher ID.
-   *                          The file portion beyond this ID is potentially
+   *                      The file portion beyond this ID is potentially being
-   *                          being updated.
+   *                      updated.
    */
-  static EditLogValidation validateEditLog(EditLogInputStream in,
+  static EditLogValidation scanEditLog(EditLogInputStream in,
-      long maxTxIdToValidate) {
+      long maxTxIdToScan) {
-    long lastPos = 0;
+    long lastPos;
     long lastTxId = HdfsServerConstants.INVALID_TXID;
     long numValid = 0;
-    FSEditLogOp op = null;
     while (true) {
+      long txid;
       lastPos = in.getPosition();
       try {
-        if ((op = in.readOp()) == null) {
+        if ((txid = in.scanNextOp()) == HdfsServerConstants.INVALID_TXID) {
           break;
         }
       } catch (Throwable t) {
-        FSImage.LOG.warn("Caught exception after reading " + numValid +
+        FSImage.LOG.warn("Caught exception after scanning through "
-            " ops from " + in + " while determining its valid length." +
+            + numValid + " ops from " + in
-            "Position was " + lastPos, t);
+            + " while determining its valid length. Position was "
+            + lastPos, t);
         in.resync();
         FSImage.LOG.warn("After resync, position is " + in.getPosition());
         continue;
       }
-      if (lastTxId == HdfsServerConstants.INVALID_TXID
+      if (lastTxId == HdfsServerConstants.INVALID_TXID || txid > lastTxId) {
-          || op.getTransactionId() > lastTxId) {
+        lastTxId = txid;
-        lastTxId = op.getTransactionId();
       }
-      if (lastTxId >= maxTxIdToValidate) {
+      if (lastTxId >= maxTxIdToScan) {
         break;
       }
-
-      numValid++;
-    }
-    return new EditLogValidation(lastPos, lastTxId, false);
-  }
-
-  static EditLogValidation scanEditLog(EditLogInputStream in) {
-    long lastPos = 0;
-    long lastTxId = HdfsServerConstants.INVALID_TXID;
-    long numValid = 0;
-    FSEditLogOp op = null;
-    while (true) {
-      lastPos = in.getPosition();
-      try {
-        if ((op = in.readOp()) == null) { // TODO
-          break;
-        }
-      } catch (Throwable t) {
-        FSImage.LOG.warn("Caught exception after reading " + numValid +
-            " ops from " + in + " while determining its valid length." +
-            "Position was " + lastPos, t);
-        in.resync();
-        FSImage.LOG.warn("After resync, position is " + in.getPosition());
-        continue;
-      }
-      if (lastTxId == HdfsServerConstants.INVALID_TXID
-          || op.getTransactionId() > lastTxId) {
-        lastTxId = op.getTransactionId();
-      }
       numValid++;
     }
     return new EditLogValidation(lastPos, lastTxId, false);

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FileJournalManager.java

@@ -212,7 +212,7 @@ public class FileJournalManager implements JournalManager {
       }
       if (elf.isInProgress()) {
         try {
-          elf.validateLog(getLastReadableTxId());
+          elf.scanLog(getLastReadableTxId(), true);
         } catch (IOException e) {
           LOG.error("got IOException while trying to validate header of " +
               elf + ".  Skipping.", e);
@@ -349,8 +349,8 @@ public class FileJournalManager implements JournalManager {
   }
   
   static void addStreamsToCollectionFromFiles(Collection<EditLogFile> elfs,
-      Collection<EditLogInputStream> streams, long fromTxId, long maxTxIdToValidate,
+      Collection<EditLogInputStream> streams, long fromTxId,
-      boolean inProgressOk) {
+      long maxTxIdToScan, boolean inProgressOk) {
     for (EditLogFile elf : elfs) {
       if (elf.isInProgress()) {
         if (!inProgressOk) {
@@ -361,7 +361,7 @@ public class FileJournalManager implements JournalManager {
           continue;
         }
         try {
-          elf.validateLog(maxTxIdToValidate);
+          elf.scanLog(maxTxIdToScan, true);
         } catch (IOException e) {
           LOG.error("got IOException while trying to validate header of " +
               elf + ".  Skipping.", e);
@@ -405,7 +405,7 @@ public class FileJournalManager implements JournalManager {
           continue;
         }
 
-        elf.validateLog(getLastReadableTxId());
+        elf.scanLog(getLastReadableTxId(), true);
 
         if (elf.hasCorruptHeader()) {
           elf.moveAsideCorruptFile();
@@ -537,20 +537,16 @@ public class FileJournalManager implements JournalManager {
      * Find out where the edit log ends.
      * This will update the lastTxId of the EditLogFile or
      * mark it as corrupt if it is.
-     * @param maxTxIdToValidate Maximum Tx ID to try to validate. Validation
+     * @param maxTxIdToScan Maximum Tx ID to try to scan.
-     *                          returns after reading this or a higher ID.
+     *                      The scan returns after reading this or a higher ID.
-     *                          The file portion beyond this ID is potentially
+     *                      The file portion beyond this ID is potentially being
-     *                          being updated.
+     *                      updated.
+     * @param verifyVersion Whether the scan should verify the layout version
      */
-    public void validateLog(long maxTxIdToValidate) throws IOException {
+    public void scanLog(long maxTxIdToScan, boolean verifyVersion)
-      EditLogValidation val = EditLogFileInputStream.validateEditLog(file,
+        throws IOException {
-          maxTxIdToValidate);
+      EditLogValidation val = EditLogFileInputStream.scanEditLog(file,
-      this.lastTxId = val.getEndTxId();
+          maxTxIdToScan, verifyVersion);
-      this.hasCorruptHeader = val.hasCorruptHeader();
-    }
-
-    public void scanLog() throws IOException {
-      EditLogValidation val = EditLogFileInputStream.scanEditLog(file);
       this.lastTxId = val.getEndTxId();
       this.hasCorruptHeader = val.hasCorruptHeader();
     }

hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/TestCheckPointForSecurityTokens.java

@@ -88,7 +88,7 @@ public class TestCheckPointForSecurityTokens {
       for (StorageDirectory sd : nn.getFSImage().getStorage().dirIterable(null)) {
         EditLogFile log = FSImageTestUtil.findLatestEditsLog(sd);
         assertTrue(log.isInProgress());
-        log.validateLog(Long.MAX_VALUE);
+        log.scanLog(Long.MAX_VALUE, true);
         long numTransactions = (log.getLastTxId() - log.getFirstTxId()) + 1;
         assertEquals("In-progress log " + log + " should have 5 transactions",
                      5, numTransactions);;
@@ -105,7 +105,7 @@ public class TestCheckPointForSecurityTokens {
       for (StorageDirectory sd : nn.getFSImage().getStorage().dirIterable(null)) {
         EditLogFile log = FSImageTestUtil.findLatestEditsLog(sd);
         assertTrue(log.isInProgress());
-        log.validateLog(Long.MAX_VALUE);
+        log.scanLog(Long.MAX_VALUE, true);
         long numTransactions = (log.getLastTxId() - log.getFirstTxId()) + 1;
         assertEquals("In-progress log " + log + " should only have START txn",
             1, numTransactions);

hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/TestEditLog.java

@@ -1229,7 +1229,7 @@ public class TestEditLog {
 
     for (EditLogInputStream edits : editStreams) {
       FSEditLogLoader.EditLogValidation val =
-          FSEditLogLoader.validateEditLog(edits, Long.MAX_VALUE);
+          FSEditLogLoader.scanEditLog(edits, Long.MAX_VALUE);
       long read = (val.getEndTxId() - edits.getFirstTxId()) + 1;
       LOG.info("Loading edits " + edits + " read " + read);
       assertEquals(startTxId, edits.getFirstTxId());

hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/TestFSEditLogLoader.java

@@ -319,7 +319,7 @@ public class TestFSEditLogLoader {
       rwf.close();
     }
     EditLogValidation validation =
-        EditLogFileInputStream.validateEditLog(logFile, Long.MAX_VALUE);
+        EditLogFileInputStream.scanEditLog(logFile, Long.MAX_VALUE, true);
     assertTrue(validation.hasCorruptHeader());
   }
 
@@ -334,7 +334,7 @@ public class TestFSEditLogLoader {
     File logFileBak = new File(testDir, logFile.getName() + ".bak");
     Files.copy(logFile, logFileBak);
     EditLogValidation validation =
-        EditLogFileInputStream.validateEditLog(logFile, Long.MAX_VALUE);
+        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.
@@ -347,8 +347,8 @@ public class TestFSEditLogLoader {
       // Restore backup, corrupt the txn opcode
       Files.copy(logFileBak, logFile);
       corruptByteInFile(logFile, txOffset);
-      validation = EditLogFileInputStream.validateEditLog(logFile,
+      validation = EditLogFileInputStream.scanEditLog(logFile,
-          Long.MAX_VALUE);
+          Long.MAX_VALUE, true);
       long expectedEndTxId = (txId == (NUM_TXNS + 1)) ?
           NUM_TXNS : (NUM_TXNS + 1);
       assertEquals("Failed when corrupting txn opcode at " + txOffset,
@@ -365,8 +365,8 @@ public class TestFSEditLogLoader {
       // Restore backup, corrupt the txn opcode
       Files.copy(logFileBak, logFile);
       truncateFile(logFile, txOffset);
-      validation = EditLogFileInputStream.validateEditLog(logFile,
+      validation = EditLogFileInputStream.scanEditLog(logFile,
-          Long.MAX_VALUE);
+          Long.MAX_VALUE, true);
       long expectedEndTxId = (txId == 0) ?
           HdfsServerConstants.INVALID_TXID : (txId - 1);
       assertEquals("Failed when corrupting txid " + txId + " txn opcode " +
@@ -384,7 +384,7 @@ public class TestFSEditLogLoader {
     // layout flags section.
     truncateFile(logFile, 8);
     EditLogValidation validation =
-        EditLogFileInputStream.validateEditLog(logFile, Long.MAX_VALUE);
+        EditLogFileInputStream.scanEditLog(logFile, Long.MAX_VALUE, true);
     assertTrue(!validation.hasCorruptHeader());
     assertEquals(HdfsServerConstants.INVALID_TXID, validation.getEndTxId());
   }