HDFS-13813. Exit NameNode if dangling child inode is detected when saving FsImage. Contributed by Siyao Meng.

(cherry picked from commit 23854443efa62aa70a1c30c32c3816750e5d7a5b)
(cherry picked from commit b38649c59a70b3112811443464b3b3180f4b9873)
(cherry picked from commit 4b606eb9fd5dde5629e791d3338886a70b5aa57c)
(cherry picked from commit ad5c3300ff7383d84ca764283e876a3889846269)

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

@@ -407,6 +407,8 @@ public final class FSImageFormatPBINode {
   }
 
   public final static class Saver {
+    private long numImageErrors;
+
     private static long buildPermissionStatus(INodeAttributes n,
         final SaverContext.DeduplicationMap<String> stringMap) {
       long userId = stringMap.getId(n.getUserName());
@@ -525,11 +527,13 @@ public final class FSImageFormatPBINode {
       this.summary = summary;
       this.context = parent.getContext();
       this.fsn = context.getSourceNamesystem();
+      this.numImageErrors = 0;
     }
 
     void serializeINodeDirectorySection(OutputStream out) throws IOException {
-      Iterator<INodeWithAdditionalFields> iter = fsn.getFSDirectory()
-          .getINodeMap().getMapIterator();
+      FSDirectory dir = fsn.getFSDirectory();
+      Iterator<INodeWithAdditionalFields> iter = dir.getINodeMap()
+          .getMapIterator();
       final ArrayList<INodeReference> refList = parent.getSaverContext()
           .getRefList();
       int i = 0;
@@ -545,6 +549,17 @@ public final class FSImageFormatPBINode {
           INodeDirectorySection.DirEntry.Builder b = INodeDirectorySection.
               DirEntry.newBuilder().setParent(n.getId());
           for (INode inode : children) {
+            // Error if the child inode doesn't exist in inodeMap
+            if (dir.getInode(inode.getId()) == null) {
+              FSImage.LOG.error(
+                  "FSImageFormatPBINode#serializeINodeDirectorySection: " +
+                      "Dangling child pointer found. Missing INode in " +
+                      "inodeMap: id=" + inode.getId() +
+                      "; path=" + inode.getFullPathName() +
+                      "; parent=" + (inode.getParent() == null ? "null" :
+                      inode.getParent().getFullPathName()));
+              ++numImageErrors;
+            }
             if (!inode.isReference()) {
               b.addChildren(inode.getId());
             } else {
@@ -672,6 +687,15 @@ public final class FSImageFormatPBINode {
           .setId(n.getId())
           .setName(ByteString.copyFrom(n.getLocalNameBytes()));
     }
+
+    /**
+     * Number of non-fatal errors detected while writing the
+     * INodeSection and INodeDirectorySection sections.
+     * @return the number of non-fatal errors detected.
+     */
+    public long getNumImageErrors() {
+      return numImageErrors;
+    }
   }
 
   private FSImageFormatPBINode() {

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

@@ -448,13 +448,15 @@ public final class FSImageFormatProtobuf {
       out.write(lengthBytes);
     }
 
-    private void saveInodes(FileSummary.Builder summary) throws IOException {
+    private long saveInodes(FileSummary.Builder summary) throws IOException {
       FSImageFormatPBINode.Saver saver = new FSImageFormatPBINode.Saver(this,
           summary);
 
       saver.serializeINodeSection(sectionOutputStream);
       saver.serializeINodeDirectorySection(sectionOutputStream);
       saver.serializeFilesUCSection(sectionOutputStream);
+
+      return saver.getNumImageErrors();
     }
 
     /**
@@ -511,8 +513,9 @@ public final class FSImageFormatProtobuf {
 
       Step step = new Step(StepType.INODES, filePath);
       prog.beginStep(Phase.SAVING_CHECKPOINT, step);
-      saveInodes(b);
-      long numErrors = saveSnapshots(b);
+      // Count number of non-fatal errors when saving inodes and snapshots.
+      long numErrors = saveInodes(b);
+      numErrors += saveSnapshots(b);
       prog.endStep(Phase.SAVING_CHECKPOINT, step);
 
       step = new Step(StepType.DELEGATION_TOKENS, filePath);