HDFS-5785. Serialize symlink in protobuf. Contributed by Haohui Mai.

git-svn-id: https://svn.apache.org/repos/asf/hadoop/common/branches/HDFS-5698@1559299 13f79535-47bb-0310-9956-ffa450edef68

hadoop-hdfs-project/hadoop-hdfs/CHANGES_HDFS-5698.txt

@@ -7,3 +7,5 @@ HDFS-5698 subtasks
     HDFS-5772. Serialize under-construction file information in FSImage. (jing9)
 
     HDFS-5783. Compute the digest before loading FSImage. (Haohui Mai via jing9)
+
+    HDFS-5785. Serialize symlink in protobuf. (Haohui Mai via jing9)

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

@@ -132,6 +132,8 @@ final class FSImageFormatPBINode {
         return loadINodeFile(n);
       case DIRECTORY:
         return loadINodeDirectory(n);
+      case SYMLINK:
+        return loadINodeSymlink(n);
       default:
         break;
       }
@@ -183,6 +185,15 @@ final class FSImageFormatPBINode {
       return file;
     }
 
+
+    private INodeSymlink loadINodeSymlink(INodeSection.INode n) {
+      assert n.getType() == INodeSection.INode.Type.SYMLINK;
+      INodeSection.INodeSymlink s = n.getSymlink();
+      final PermissionStatus permissions = loadPermission(s.getPermission());
+      return new INodeSymlink(n.getId(), n.getName().toByteArray(), permissions,
+          0, 0, s.getTarget().toStringUtf8());
+    }
+
     private void loadRootINode(INodeSection.INode p) {
       INodeDirectory root = loadINodeDirectory(p);
       final Quota.Counts q = root.getQuotaCounts();
@@ -277,6 +288,8 @@ final class FSImageFormatPBINode {
         save(out, n.asDirectory());
       } else if (n.isFile()) {
         save(out, n.asFile());
+      } else if (n.isSymlink()) {
+        save(out, n.asSymlink());
       }
     }
 
@@ -288,9 +301,8 @@ final class FSImageFormatPBINode {
           .setDsQuota(quota.get(Quota.DISKSPACE))
           .setPermission(buildPermissionStatus(n));
 
-      INodeSection.INode r = INodeSection.INode.newBuilder()
-          .setType(INodeSection.INode.Type.DIRECTORY).setId(n.getId())
-          .setName(ByteString.copyFrom(n.getLocalNameBytes())).setDirectory(b).build();
+      INodeSection.INode r = buildINodeCommon(n)
+          .setType(INodeSection.INode.Type.DIRECTORY).setDirectory(b).build();
       r.writeDelimitedTo(out);
     }
 
@@ -314,11 +326,25 @@ final class FSImageFormatPBINode {
         b.setFileUC(f);
       }
 
-      INodeSection.INode r = INodeSection.INode.newBuilder()
-          .setType(INodeSection.INode.Type.FILE).setId(n.getId())
-          .setName(ByteString.copyFrom(n.getLocalNameBytes())).setFile(b).build();
+      INodeSection.INode r = buildINodeCommon(n)
+          .setType(INodeSection.INode.Type.FILE).setFile(b).build();
+      r.writeDelimitedTo(out);
+    }
+
+    private void save(OutputStream out, INodeSymlink n) throws IOException {
+      INodeSection.INodeSymlink.Builder b = INodeSection.INodeSymlink
+          .newBuilder().setPermission(buildPermissionStatus(n))
+          .setTarget(ByteString.copyFrom(n.getSymlink()));
+      INodeSection.INode r = buildINodeCommon(n)
+          .setType(INodeSection.INode.Type.SYMLINK).setSymlink(b).build();
       r.writeDelimitedTo(out);
     }
+
+    private final INodeSection.INode.Builder buildINodeCommon(INode n) {
+      return INodeSection.INode.newBuilder()
+          .setId(n.getId())
+          .setName(ByteString.copyFrom(n.getLocalNameBytes()));
+    }
   }
 
   private FSImageFormatPBINode() {

hadoop-hdfs-project/hadoop-hdfs/src/main/proto/fsimage.proto

@@ -109,12 +109,16 @@ message INodeSection {
     optional Permission permission = 4;
   }
 
+  message INodeSymlink {
+    optional Permission permission = 1;
+    optional bytes target = 2;
+  }
+
   message INode {
     enum Type {
       FILE = 1;
       DIRECTORY = 2;
-      REFERENCE = 3;
-      SYMLINK = 4;
+      SYMLINK = 3;
     };
     required Type type = 1;
     required uint64 id = 2;
@@ -122,6 +126,7 @@ message INodeSection {
     
     optional INodeFile file = 4;
     optional INodeDirectory directory = 5;
+    optional INodeSymlink symlink = 6;
   }
 
   optional uint64 lastInodeId = 1;