HDFS-6414. xattr modification operations are based on state of latest snapshot instead of current version of inode. Contributed by Andrew Wang.

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

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

@@ -64,3 +64,6 @@ HDFS-2006 (Unreleased)
 
     HDFS-6413. xattr names erroneously handled as case-insensitive.
     (Charles Lamb via cnauroth)
+
+    HDFS-6414. xattr modification operations are based on state of latest
+    snapshot instead of current version of inode. (Andrew Wang via cnauroth)

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

@@ -60,21 +60,21 @@ public class XAttrStorage {
    * Update xattrs of inode.
    * @param inode INode to update
    * @param xAttrs to update xAttrs.
-   * @param snapshotId
+   * @param snapshotId id of the latest snapshot of the inode
    */
   public static void updateINodeXAttrs(INode inode, 
       List<XAttr> xAttrs, int snapshotId) throws QuotaExceededException {
     if (xAttrs == null || xAttrs.isEmpty()) {
-      if (inode.getXAttrFeature(snapshotId) != null) {
+      if (inode.getXAttrFeature() != null) {
         inode.removeXAttrFeature(snapshotId);
       }
       return;
     }
     
     ImmutableList<XAttr> newXAttrs = ImmutableList.copyOf(xAttrs);
-    if (inode.getXAttrFeature(snapshotId) != null) {
+    if (inode.getXAttrFeature() != null) {
       inode.removeXAttrFeature(snapshotId);
     }
-    inode.addXAttrFeature(new XAttrFeature(newXAttrs));
+    inode.addXAttrFeature(new XAttrFeature(newXAttrs), snapshotId);
   }
 }

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

@@ -18,11 +18,16 @@
 
 package org.apache.hadoop.hdfs.server.namenode.snapshot;
 
+import static org.junit.Assert.assertArrayEquals;
+import static org.junit.Assert.assertEquals;
+
+import java.util.EnumSet;
 import java.util.Map;
 
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.XAttrSetFlag;
 import org.apache.hadoop.fs.permission.FsPermission;
 import org.apache.hadoop.hdfs.DFSConfigKeys;
 import org.apache.hadoop.hdfs.DistributedFileSystem;
@@ -85,6 +90,80 @@ public class TestXAttrWithSnapshot {
     snapshotPath = new Path(path, new Path(".snapshot", snapshotName));
   }
 
+  /**
+   * Tests modifying xattrs on a directory that has been snapshotted
+   */
+  @Test (timeout = 120000)
+  public void testModifyReadsCurrentState() throws Exception {
+    // Init
+    FileSystem.mkdirs(hdfs, path, FsPermission.createImmutable((short) 0700));
+    SnapshotTestHelper.createSnapshot(hdfs, path, snapshotName);
+    hdfs.setXAttr(path, name1, value1);
+    hdfs.setXAttr(path, name2, value2);
+
+    // Verify that current path reflects xattrs, snapshot doesn't
+    Map<String, byte[]> xattrs = hdfs.getXAttrs(path);
+    assertEquals(xattrs.size(), 2);
+    assertArrayEquals(value1, xattrs.get(name1));
+    assertArrayEquals(value2, xattrs.get(name2));
+
+    xattrs = hdfs.getXAttrs(snapshotPath);
+    assertEquals(xattrs.size(), 0);
+
+    // Modify each xattr and make sure it's reflected
+    hdfs.setXAttr(path, name1, value2, EnumSet.of(XAttrSetFlag.REPLACE));
+    xattrs = hdfs.getXAttrs(path);
+    assertEquals(xattrs.size(), 2);
+    assertArrayEquals(value2, xattrs.get(name1));
+    assertArrayEquals(value2, xattrs.get(name2));
+
+    hdfs.setXAttr(path, name2, value1, EnumSet.of(XAttrSetFlag.REPLACE));
+    xattrs = hdfs.getXAttrs(path);
+    assertEquals(xattrs.size(), 2);
+    assertArrayEquals(value2, xattrs.get(name1));
+    assertArrayEquals(value1, xattrs.get(name2));
+
+    // Paranoia checks
+    xattrs = hdfs.getXAttrs(snapshotPath);
+    assertEquals(xattrs.size(), 0);
+
+    hdfs.removeXAttr(path, name1);
+    hdfs.removeXAttr(path, name2);
+    xattrs = hdfs.getXAttrs(path);
+    assertEquals(xattrs.size(), 0);
+  }
+
+  /**
+   * Tests removing xattrs on a directory that has been snapshotted
+   */
+  @Test (timeout = 120000)
+  public void testRemoveReadsCurrentState() throws Exception {
+    // Init
+    FileSystem.mkdirs(hdfs, path, FsPermission.createImmutable((short) 0700));
+    SnapshotTestHelper.createSnapshot(hdfs, path, snapshotName);
+    hdfs.setXAttr(path, name1, value1);
+    hdfs.setXAttr(path, name2, value2);
+
+    // Verify that current path reflects xattrs, snapshot doesn't
+    Map<String, byte[]> xattrs = hdfs.getXAttrs(path);
+    assertEquals(xattrs.size(), 2);
+    assertArrayEquals(value1, xattrs.get(name1));
+    assertArrayEquals(value2, xattrs.get(name2));
+
+    xattrs = hdfs.getXAttrs(snapshotPath);
+    assertEquals(xattrs.size(), 0);
+
+    // Remove xattrs and verify one-by-one
+    hdfs.removeXAttr(path, name2);
+    xattrs = hdfs.getXAttrs(path);
+    assertEquals(xattrs.size(), 1);
+    assertArrayEquals(value1, xattrs.get(name1));
+
+    hdfs.removeXAttr(path, name1);
+    xattrs = hdfs.getXAttrs(path);
+    assertEquals(xattrs.size(), 0);
+  }
+
   /**
    * 1) Save xattrs, then create snapshot. Assert that inode of original and
    * snapshot have same xattrs. 2) Change the original xattrs, assert snapshot
@@ -215,6 +294,25 @@ public class TestXAttrWithSnapshot {
     hdfs.setXAttr(filePath, name2, value2);
   }
 
+
+  /**
+   * Test that an exception is thrown when adding an XAttr Feature to
+   * a snapshotted path
+   */
+  @Test
+  public void testSetXAttrAfterSnapshotExceedsQuota() throws Exception {
+    Path filePath = new Path(path, "file1");
+    FileSystem.mkdirs(hdfs, path, FsPermission.createImmutable((short) 0755));
+    hdfs.allowSnapshot(path);
+    hdfs.setQuota(path, 3, HdfsConstants.QUOTA_DONT_SET);
+    FileSystem.create(hdfs, filePath,
+        FsPermission.createImmutable((short) 0600)).close();
+    hdfs.createSnapshot(path, snapshotName);
+    // This adds an XAttr feature, which can throw an exception
+    exception.expect(NSQuotaExceededException.class);
+    hdfs.setXAttr(filePath, name1, value1);
+  }
+
   /**
    * Assert exception of removing xattr when exceeding quota.
    */