Java 类org.apache.hadoop.hdfs.server.namenode.snapshot.DirectoryWithSnapshotFeature.DirectoryDiff 实例源码

/**
 * Load a node stored in the created list from fsimage.
 * @param createdNodeName The name of the created node.
 * @param parent The directory that the created list belongs to.
 * @return The created node.
 */
public static INode loadCreated(byte[] createdNodeName,
    INodeDirectory parent) throws IOException {
  // the INode in the created list should be a reference to another INode
  // in posterior SnapshotDiffs or one of the current children
  for (DirectoryDiff postDiff : parent.getDiffs()) {
    final INode d = postDiff.getChildrenDiff().search(ListType.DELETED,
        createdNodeName);
    if (d != null) {
      return d;
    } // else go to the next SnapshotDiff
  } 
  // use the current child
  INode currentChild = parent.getChild(createdNodeName,
      Snapshot.CURRENT_STATE_ID);
  if (currentChild == null) {
    throw new IOException("Cannot find an INode associated with the INode "
        + DFSUtil.bytes2String(createdNodeName)
        + " in created list while loading FSImage.");
  }
  return currentChild;
}

/**
 * Load {@link DirectoryDiff} from fsimage.
 * @param parent The directory that the SnapshotDiff belongs to.
 * @param in The {@link DataInput} instance to read.
 * @param loader The {@link Loader} instance that this loading procedure is 
 *               using.
 * @return A {@link DirectoryDiff}.
 */
private static DirectoryDiff loadDirectoryDiff(INodeDirectory parent,
    DataInput in, FSImageFormat.Loader loader) throws IOException {
  // 1. Read the full path of the Snapshot root to identify the Snapshot
  final Snapshot snapshot = loader.getSnapshot(in);

  // 2. Load DirectoryDiff#childrenSize
  int childrenSize = in.readInt();

  // 3. Load DirectoryDiff#snapshotINode 
  INodeDirectoryAttributes snapshotINode = loadSnapshotINodeInDirectoryDiff(
      snapshot, in, loader);

  // 4. Load the created list in SnapshotDiff#Diff
  List<INode> createdList = loadCreatedList(parent, in);

  // 5. Load the deleted list in SnapshotDiff#Diff
  List<INode> deletedList = loadDeletedList(parent, createdList, in, loader);

  // 6. Compose the SnapshotDiff
  List<DirectoryDiff> diffs = parent.getDiffs().asList();
  DirectoryDiff sdiff = new DirectoryDiff(snapshot.getId(), snapshotINode,
      diffs.isEmpty() ? null : diffs.get(0), childrenSize, createdList,
      deletedList, snapshotINode == snapshot.getRoot());
  return sdiff;
}

/**
 * Check the correctness of snapshot list within snapshottable dir
 */
private void checkSnapshotList(INodeDirectory srcRoot,
    String[] sortedNames, String[] names) {
  assertTrue(srcRoot.isSnapshottable());
  ReadOnlyList<Snapshot> listByName = srcRoot
      .getDirectorySnapshottableFeature().getSnapshotList();
  assertEquals(sortedNames.length, listByName.size());
  for (int i = 0; i < listByName.size(); i++) {
    assertEquals(sortedNames[i], listByName.get(i).getRoot().getLocalName());
  }
  List<DirectoryDiff> listByTime = srcRoot.getDiffs().asList();
  assertEquals(names.length, listByTime.size());
  for (int i = 0; i < listByTime.size(); i++) {
    Snapshot s = srcRoot.getDirectorySnapshottableFeature().getSnapshotById(
        listByTime.get(i).getSnapshotId());
    assertEquals(names[i], s.getRoot().getLocalName());
  }
}

/**
 * Load a node stored in the created list from fsimage.
 * @param createdNodeName The name of the created node.
 * @param parent The directory that the created list belongs to.
 * @return The created node.
 */
public static INode loadCreated(byte[] createdNodeName,
    INodeDirectory parent) throws IOException {
  // the INode in the created list should be a reference to another INode
  // in posterior SnapshotDiffs or one of the current children
  for (DirectoryDiff postDiff : parent.getDiffs()) {
    final INode d = postDiff.getChildrenDiff().search(ListType.DELETED,
        createdNodeName);
    if (d != null) {
      return d;
    } // else go to the next SnapshotDiff
  } 
  // use the current child
  INode currentChild = parent.getChild(createdNodeName,
      Snapshot.CURRENT_STATE_ID);
  if (currentChild == null) {
    throw new IOException("Cannot find an INode associated with the INode "
        + DFSUtil.bytes2String(createdNodeName)
        + " in created list while loading FSImage.");
  }
  return currentChild;
}

/**
 * Load {@link DirectoryDiff} from fsimage.
 * @param parent The directory that the SnapshotDiff belongs to.
 * @param in The {@link DataInput} instance to read.
 * @param loader The {@link Loader} instance that this loading procedure is 
 *               using.
 * @return A {@link DirectoryDiff}.
 */
private static DirectoryDiff loadDirectoryDiff(INodeDirectory parent,
    DataInput in, FSImageFormat.Loader loader) throws IOException {
  // 1. Read the full path of the Snapshot root to identify the Snapshot
  final Snapshot snapshot = loader.getSnapshot(in);

  // 2. Load DirectoryDiff#childrenSize
  int childrenSize = in.readInt();

  // 3. Load DirectoryDiff#snapshotINode 
  INodeDirectoryAttributes snapshotINode = loadSnapshotINodeInDirectoryDiff(
      snapshot, in, loader);

  // 4. Load the created list in SnapshotDiff#Diff
  List<INode> createdList = loadCreatedList(parent, in);

  // 5. Load the deleted list in SnapshotDiff#Diff
  List<INode> deletedList = loadDeletedList(parent, createdList, in, loader);

  // 6. Compose the SnapshotDiff
  List<DirectoryDiff> diffs = parent.getDiffs().asList();
  DirectoryDiff sdiff = new DirectoryDiff(snapshot.getId(), snapshotINode,
      diffs.isEmpty() ? null : diffs.get(0), childrenSize, createdList,
      deletedList, snapshotINode == snapshot.getRoot());
  return sdiff;
}

/**
 * Check the correctness of snapshot list within snapshottable dir
 */
private void checkSnapshotList(INodeDirectory srcRoot,
    String[] sortedNames, String[] names) {
  assertTrue(srcRoot.isSnapshottable());
  ReadOnlyList<Snapshot> listByName = srcRoot
      .getDirectorySnapshottableFeature().getSnapshotList();
  assertEquals(sortedNames.length, listByName.size());
  for (int i = 0; i < listByName.size(); i++) {
    assertEquals(sortedNames[i], listByName.get(i).getRoot().getLocalName());
  }
  List<DirectoryDiff> listByTime = srcRoot.getDiffs().asList();
  assertEquals(names.length, listByTime.size());
  for (int i = 0; i < listByTime.size(); i++) {
    Snapshot s = srcRoot.getDirectorySnapshottableFeature().getSnapshotById(
        listByTime.get(i).getSnapshotId());
    assertEquals(names[i], s.getRoot().getLocalName());
  }
}

/**
 * Load a node stored in the created list from fsimage.
 * @param createdNodeName The name of the created node.
 * @param parent The directory that the created list belongs to.
 * @return The created node.
 */
public static INode loadCreated(byte[] createdNodeName,
    INodeDirectory parent) throws IOException {
  // the INode in the created list should be a reference to another INode
  // in posterior SnapshotDiffs or one of the current children
  for (DirectoryDiff postDiff : parent.getDiffs()) {
    final INode d = postDiff.getChildrenDiff().search(ListType.DELETED,
        createdNodeName);
    if (d != null) {
      return d;
    } // else go to the next SnapshotDiff
  } 
  // use the current child
  INode currentChild = parent.getChild(createdNodeName,
      Snapshot.CURRENT_STATE_ID);
  if (currentChild == null) {
    throw new IOException("Cannot find an INode associated with the INode "
        + DFSUtil.bytes2String(createdNodeName)
        + " in created list while loading FSImage.");
  }
  return currentChild;
}

/**
 * Load {@link DirectoryDiff} from fsimage.
 * @param parent The directory that the SnapshotDiff belongs to.
 * @param in The {@link DataInput} instance to read.
 * @param loader The {@link Loader} instance that this loading procedure is 
 *               using.
 * @return A {@link DirectoryDiff}.
 */
private static DirectoryDiff loadDirectoryDiff(INodeDirectory parent,
    DataInput in, FSImageFormat.Loader loader) throws IOException {
  // 1. Read the full path of the Snapshot root to identify the Snapshot
  final Snapshot snapshot = loader.getSnapshot(in);

  // 2. Load DirectoryDiff#childrenSize
  int childrenSize = in.readInt();

  // 3. Load DirectoryDiff#snapshotINode 
  INodeDirectoryAttributes snapshotINode = loadSnapshotINodeInDirectoryDiff(
      snapshot, in, loader);

  // 4. Load the created list in SnapshotDiff#Diff
  List<INode> createdList = loadCreatedList(parent, in);

  // 5. Load the deleted list in SnapshotDiff#Diff
  List<INode> deletedList = loadDeletedList(parent, createdList, in, loader);

  // 6. Compose the SnapshotDiff
  List<DirectoryDiff> diffs = parent.getDiffs().asList();
  DirectoryDiff sdiff = new DirectoryDiff(snapshot.getId(), snapshotINode,
      diffs.isEmpty() ? null : diffs.get(0), childrenSize, createdList,
      deletedList, snapshotINode == snapshot.getRoot());
  return sdiff;
}

/**
 * Check the correctness of snapshot list within snapshottable dir
 */
private void checkSnapshotList(INodeDirectory srcRoot,
    String[] sortedNames, String[] names) {
  assertTrue(srcRoot.isSnapshottable());
  ReadOnlyList<Snapshot> listByName = srcRoot
      .getDirectorySnapshottableFeature().getSnapshotList();
  assertEquals(sortedNames.length, listByName.size());
  for (int i = 0; i < listByName.size(); i++) {
    assertEquals(sortedNames[i], listByName.get(i).getRoot().getLocalName());
  }
  List<DirectoryDiff> listByTime = srcRoot.getDiffs().asList();
  assertEquals(names.length, listByTime.size());
  for (int i = 0; i < listByTime.size(); i++) {
    Snapshot s = srcRoot.getDirectorySnapshottableFeature().getSnapshotById(
        listByTime.get(i).getSnapshotId());
    assertEquals(names[i], s.getRoot().getLocalName());
  }
}

/**
 * Load a node stored in the created list from fsimage.
 * @param createdNodeName The name of the created node.
 * @param parent The directory that the created list belongs to.
 * @return The created node.
 */
public static INode loadCreated(byte[] createdNodeName,
    INodeDirectory parent) throws IOException {
  // the INode in the created list should be a reference to another INode
  // in posterior SnapshotDiffs or one of the current children
  for (DirectoryDiff postDiff : parent.getDiffs()) {
    final INode d = postDiff.getChildrenDiff().search(ListType.DELETED,
        createdNodeName);
    if (d != null) {
      return d;
    } // else go to the next SnapshotDiff
  } 
  // use the current child
  INode currentChild = parent.getChild(createdNodeName,
      Snapshot.CURRENT_STATE_ID);
  if (currentChild == null) {
    throw new IOException("Cannot find an INode associated with the INode "
        + DFSUtil.bytes2String(createdNodeName)
        + " in created list while loading FSImage.");
  }
  return currentChild;
}

/**
 * Load {@link DirectoryDiff} from fsimage.
 * @param parent The directory that the SnapshotDiff belongs to.
 * @param in The {@link DataInput} instance to read.
 * @param loader The {@link Loader} instance that this loading procedure is 
 *               using.
 * @return A {@link DirectoryDiff}.
 */
private static DirectoryDiff loadDirectoryDiff(INodeDirectory parent,
    DataInput in, FSImageFormat.Loader loader) throws IOException {
  // 1. Read the full path of the Snapshot root to identify the Snapshot
  final Snapshot snapshot = loader.getSnapshot(in);

  // 2. Load DirectoryDiff#childrenSize
  int childrenSize = in.readInt();

  // 3. Load DirectoryDiff#snapshotINode 
  INodeDirectoryAttributes snapshotINode = loadSnapshotINodeInDirectoryDiff(
      snapshot, in, loader);

  // 4. Load the created list in SnapshotDiff#Diff
  List<INode> createdList = loadCreatedList(parent, in);

  // 5. Load the deleted list in SnapshotDiff#Diff
  List<INode> deletedList = loadDeletedList(parent, createdList, in, loader);

  // 6. Compose the SnapshotDiff
  List<DirectoryDiff> diffs = parent.getDiffs().asList();
  DirectoryDiff sdiff = new DirectoryDiff(snapshot.getId(), snapshotINode,
      diffs.isEmpty() ? null : diffs.get(0), childrenSize, createdList,
      deletedList, snapshotINode == snapshot.getRoot());
  return sdiff;
}

/**
 * Check the correctness of snapshot list within snapshottable dir
 */
private void checkSnapshotList(INodeDirectory srcRoot,
    String[] sortedNames, String[] names) {
  assertTrue(srcRoot.isSnapshottable());
  ReadOnlyList<Snapshot> listByName = srcRoot
      .getDirectorySnapshottableFeature().getSnapshotList();
  assertEquals(sortedNames.length, listByName.size());
  for (int i = 0; i < listByName.size(); i++) {
    assertEquals(sortedNames[i], listByName.get(i).getRoot().getLocalName());
  }
  List<DirectoryDiff> listByTime = srcRoot.getDiffs().asList();
  assertEquals(names.length, listByTime.size());
  for (int i = 0; i < listByTime.size(); i++) {
    Snapshot s = srcRoot.getDirectorySnapshottableFeature().getSnapshotById(
        listByTime.get(i).getSnapshotId());
    assertEquals(names[i], s.getRoot().getLocalName());
  }
}

/**
 * Load a node stored in the created list from fsimage.
 * @param createdNodeName The name of the created node.
 * @param parent The directory that the created list belongs to.
 * @return The created node.
 */
public static INode loadCreated(byte[] createdNodeName,
    INodeDirectory parent) throws IOException {
  // the INode in the created list should be a reference to another INode
  // in posterior SnapshotDiffs or one of the current children
  for (DirectoryDiff postDiff : parent.getDiffs()) {
    final INode d = postDiff.getChildrenDiff().search(ListType.DELETED,
        createdNodeName);
    if (d != null) {
      return d;
    } // else go to the next SnapshotDiff
  } 
  // use the current child
  INode currentChild = parent.getChild(createdNodeName,
      Snapshot.CURRENT_STATE_ID);
  if (currentChild == null) {
    throw new IOException("Cannot find an INode associated with the INode "
        + DFSUtil.bytes2String(createdNodeName)
        + " in created list while loading FSImage.");
  }
  return currentChild;
}

/**
 * Load {@link DirectoryDiff} from fsimage.
 * @param parent The directory that the SnapshotDiff belongs to.
 * @param in The {@link DataInput} instance to read.
 * @param loader The {@link Loader} instance that this loading procedure is 
 *               using.
 * @return A {@link DirectoryDiff}.
 */
private static DirectoryDiff loadDirectoryDiff(INodeDirectory parent,
    DataInput in, FSImageFormat.Loader loader) throws IOException {
  // 1. Read the full path of the Snapshot root to identify the Snapshot
  final Snapshot snapshot = loader.getSnapshot(in);

  // 2. Load DirectoryDiff#childrenSize
  int childrenSize = in.readInt();

  // 3. Load DirectoryDiff#snapshotINode 
  INodeDirectoryAttributes snapshotINode = loadSnapshotINodeInDirectoryDiff(
      snapshot, in, loader);

  // 4. Load the created list in SnapshotDiff#Diff
  List<INode> createdList = loadCreatedList(parent, in);

  // 5. Load the deleted list in SnapshotDiff#Diff
  List<INode> deletedList = loadDeletedList(parent, createdList, in, loader);

  // 6. Compose the SnapshotDiff
  List<DirectoryDiff> diffs = parent.getDiffs().asList();
  DirectoryDiff sdiff = new DirectoryDiff(snapshot.getId(), snapshotINode,
      diffs.isEmpty() ? null : diffs.get(0), childrenSize, createdList,
      deletedList, snapshotINode == snapshot.getRoot());
  return sdiff;
}

/**
 * Check the correctness of snapshot list within snapshottable dir
 */
private void checkSnapshotList(INodeDirectory srcRoot,
    String[] sortedNames, String[] names) {
  assertTrue(srcRoot.isSnapshottable());
  ReadOnlyList<Snapshot> listByName = srcRoot
      .getDirectorySnapshottableFeature().getSnapshotList();
  assertEquals(sortedNames.length, listByName.size());
  for (int i = 0; i < listByName.size(); i++) {
    assertEquals(sortedNames[i], listByName.get(i).getRoot().getLocalName());
  }
  List<DirectoryDiff> listByTime = srcRoot.getDiffs().asList();
  assertEquals(names.length, listByTime.size());
  for (int i = 0; i < listByTime.size(); i++) {
    Snapshot s = srcRoot.getDirectorySnapshottableFeature().getSnapshotById(
        listByTime.get(i).getSnapshotId());
    assertEquals(names[i], s.getRoot().getLocalName());
  }
}

/**
 * Load a node stored in the created list from fsimage.
 * @param createdNodeName The name of the created node.
 * @param parent The directory that the created list belongs to.
 * @return The created node.
 */
public static INode loadCreated(byte[] createdNodeName,
    INodeDirectory parent) throws IOException {
  // the INode in the created list should be a reference to another INode
  // in posterior SnapshotDiffs or one of the current children
  for (DirectoryDiff postDiff : parent.getDiffs()) {
    final INode d = postDiff.getChildrenDiff().search(ListType.DELETED,
        createdNodeName);
    if (d != null) {
      return d;
    } // else go to the next SnapshotDiff
  } 
  // use the current child
  INode currentChild = parent.getChild(createdNodeName,
      Snapshot.CURRENT_STATE_ID);
  if (currentChild == null) {
    throw new IOException("Cannot find an INode associated with the INode "
        + DFSUtil.bytes2String(createdNodeName)
        + " in created list while loading FSImage.");
  }
  return currentChild;
}

/**
 * Load {@link DirectoryDiff} from fsimage.
 * @param parent The directory that the SnapshotDiff belongs to.
 * @param in The {@link DataInput} instance to read.
 * @param loader The {@link Loader} instance that this loading procedure is 
 *               using.
 * @return A {@link DirectoryDiff}.
 */
private static DirectoryDiff loadDirectoryDiff(INodeDirectory parent,
    DataInput in, FSImageFormat.Loader loader) throws IOException {
  // 1. Read the full path of the Snapshot root to identify the Snapshot
  final Snapshot snapshot = loader.getSnapshot(in);

  // 2. Load DirectoryDiff#childrenSize
  int childrenSize = in.readInt();

  // 3. Load DirectoryDiff#snapshotINode 
  INodeDirectoryAttributes snapshotINode = loadSnapshotINodeInDirectoryDiff(
      snapshot, in, loader);

  // 4. Load the created list in SnapshotDiff#Diff
  List<INode> createdList = loadCreatedList(parent, in);

  // 5. Load the deleted list in SnapshotDiff#Diff
  List<INode> deletedList = loadDeletedList(parent, createdList, in, loader);

  // 6. Compose the SnapshotDiff
  List<DirectoryDiff> diffs = parent.getDiffs().asList();
  DirectoryDiff sdiff = new DirectoryDiff(snapshot.getId(), snapshotINode,
      diffs.isEmpty() ? null : diffs.get(0), childrenSize, createdList,
      deletedList, snapshotINode == snapshot.getRoot());
  return sdiff;
}

private void serializeDirDiffList(INodeDirectory dir,
    final List<INodeReference> refList, OutputStream out)
    throws IOException {
  DirectoryWithSnapshotFeature sf = dir.getDirectoryWithSnapshotFeature();
  if (sf != null) {
    List<DirectoryDiff> diffList = sf.getDiffs().asList();
    SnapshotDiffSection.DiffEntry entry = SnapshotDiffSection.DiffEntry
        .newBuilder().setInodeId(dir.getId()).setType(Type.DIRECTORYDIFF)
        .setNumOfDiff(diffList.size()).build();
    entry.writeDelimitedTo(out);
    for (int i = diffList.size() - 1; i >= 0; i--) { // reverse order!
      DirectoryDiff diff = diffList.get(i);
      SnapshotDiffSection.DirectoryDiff.Builder db = SnapshotDiffSection.
          DirectoryDiff.newBuilder().setSnapshotId(diff.getSnapshotId())
                       .setChildrenSize(diff.getChildrenSize())
                       .setIsSnapshotRoot(diff.isSnapshotRoot());
      INodeDirectoryAttributes copy = diff.snapshotINode;
      if (!diff.isSnapshotRoot() && copy != null) {
        db.setName(ByteString.copyFrom(copy.getLocalNameBytes()))
            .setSnapshotCopy(
                buildINodeDirectory(copy, parent.getSaverContext()));
      }
      // process created list and deleted list
      List<INode> created = diff.getChildrenDiff()
          .getList(ListType.CREATED);
      db.setCreatedListSize(created.size());
      List<INode> deleted = diff.getChildrenDiff().getList(ListType.DELETED);
      for (INode d : deleted) {
        if (d.isReference()) {
          refList.add(d.asReference());
          db.addDeletedINodeRef(refList.size() - 1);
        } else {
          db.addDeletedINode(d.getId());
        }
      }
      db.build().writeDelimitedTo(out);
      saveCreatedList(created, out);
    }
  }
}

/**
 * Make sure we clean the whole subtree under a DstReference node after 
 * deleting a snapshot.
 * see HDFS-5476.
 */
@Test
public void testCleanDstReference() throws Exception {
  final Path test = new Path("/test");
  final Path foo = new Path(test, "foo");
  final Path bar = new Path(foo, "bar");
  hdfs.mkdirs(bar);
  SnapshotTestHelper.createSnapshot(hdfs, test, "s0");

  // create file after s0 so that the file should not be included in s0
  final Path fileInBar = new Path(bar, "file");
  DFSTestUtil.createFile(hdfs, fileInBar, BLOCKSIZE, REPL, SEED);
  // rename foo --> foo2
  final Path foo2 = new Path(test, "foo2");
  hdfs.rename(foo, foo2);
  // create snapshot s1, note the file is included in s1
  hdfs.createSnapshot(test, "s1");
  // delete bar and foo2
  hdfs.delete(new Path(foo2, "bar"), true);
  hdfs.delete(foo2, true);

  final Path sfileInBar = SnapshotTestHelper.getSnapshotPath(test, "s1",
      "foo2/bar/file");
  assertTrue(hdfs.exists(sfileInBar));

  hdfs.deleteSnapshot(test, "s1");
  assertFalse(hdfs.exists(sfileInBar));

  restartClusterAndCheckImage(true);
  // make sure the file under bar is deleted 
  final Path barInS0 = SnapshotTestHelper.getSnapshotPath(test, "s0",
      "foo/bar");
  INodeDirectory barNode = fsdir.getINode(barInS0.toString()).asDirectory();
  assertEquals(0, barNode.getChildrenList(Snapshot.CURRENT_STATE_ID).size());
  List<DirectoryDiff> diffList = barNode.getDiffs().asList();
  assertEquals(1, diffList.size());
  DirectoryDiff diff = diffList.get(0);
  assertEquals(0, diff.getChildrenDiff().getList(ListType.DELETED).size());
  assertEquals(0, diff.getChildrenDiff().getList(ListType.CREATED).size());
}

private void serializeDirDiffList(INodeDirectory dir,
    final List<INodeReference> refList, OutputStream out)
    throws IOException {
  DirectoryWithSnapshotFeature sf = dir.getDirectoryWithSnapshotFeature();
  if (sf != null) {
    List<DirectoryDiff> diffList = sf.getDiffs().asList();
    SnapshotDiffSection.DiffEntry entry = SnapshotDiffSection.DiffEntry
        .newBuilder().setInodeId(dir.getId()).setType(Type.DIRECTORYDIFF)
        .setNumOfDiff(diffList.size()).build();
    entry.writeDelimitedTo(out);
    for (int i = diffList.size() - 1; i >= 0; i--) { // reverse order!
      DirectoryDiff diff = diffList.get(i);
      SnapshotDiffSection.DirectoryDiff.Builder db = SnapshotDiffSection.
          DirectoryDiff.newBuilder().setSnapshotId(diff.getSnapshotId())
                       .setChildrenSize(diff.getChildrenSize())
                       .setIsSnapshotRoot(diff.isSnapshotRoot());
      INodeDirectoryAttributes copy = diff.snapshotINode;
      if (!diff.isSnapshotRoot() && copy != null) {
        db.setName(ByteString.copyFrom(copy.getLocalNameBytes()))
            .setSnapshotCopy(
                buildINodeDirectory(copy, parent.getSaverContext()));
      }
      // process created list and deleted list
      List<INode> created = diff.getChildrenDiff()
          .getList(ListType.CREATED);
      db.setCreatedListSize(created.size());
      List<INode> deleted = diff.getChildrenDiff().getList(ListType.DELETED);
      for (INode d : deleted) {
        if (d.isReference()) {
          refList.add(d.asReference());
          db.addDeletedINodeRef(refList.size() - 1);
        } else {
          db.addDeletedINode(d.getId());
        }
      }
      db.build().writeDelimitedTo(out);
      saveCreatedList(created, out);
    }
  }
}

/**
 * Make sure we clean the whole subtree under a DstReference node after 
 * deleting a snapshot.
 * see HDFS-5476.
 */
@Test
public void testCleanDstReference() throws Exception {
  final Path test = new Path("/test");
  final Path foo = new Path(test, "foo");
  final Path bar = new Path(foo, "bar");
  hdfs.mkdirs(bar);
  SnapshotTestHelper.createSnapshot(hdfs, test, "s0");

  // create file after s0 so that the file should not be included in s0
  final Path fileInBar = new Path(bar, "file");
  DFSTestUtil.createFile(hdfs, fileInBar, BLOCKSIZE, REPL, SEED);
  // rename foo --> foo2
  final Path foo2 = new Path(test, "foo2");
  hdfs.rename(foo, foo2);
  // create snapshot s1, note the file is included in s1
  hdfs.createSnapshot(test, "s1");
  // delete bar and foo2
  hdfs.delete(new Path(foo2, "bar"), true);
  hdfs.delete(foo2, true);

  final Path sfileInBar = SnapshotTestHelper.getSnapshotPath(test, "s1",
      "foo2/bar/file");
  assertTrue(hdfs.exists(sfileInBar));

  hdfs.deleteSnapshot(test, "s1");
  assertFalse(hdfs.exists(sfileInBar));

  restartClusterAndCheckImage(true);
  // make sure the file under bar is deleted 
  final Path barInS0 = SnapshotTestHelper.getSnapshotPath(test, "s0",
      "foo/bar");
  INodeDirectory barNode = fsdir.getINode(barInS0.toString()).asDirectory();
  assertEquals(0, barNode.getChildrenList(Snapshot.CURRENT_STATE_ID).size());
  List<DirectoryDiff> diffList = barNode.getDiffs().asList();
  assertEquals(1, diffList.size());
  DirectoryDiff diff = diffList.get(0);
  assertEquals(0, diff.getChildrenDiff().getList(ListType.DELETED).size());
  assertEquals(0, diff.getChildrenDiff().getList(ListType.CREATED).size());
}

private void serializeDirDiffList(INodeDirectory dir,
    final List<INodeReference> refList, OutputStream out)
    throws IOException {
  DirectoryWithSnapshotFeature sf = dir.getDirectoryWithSnapshotFeature();
  if (sf != null) {
    List<DirectoryDiff> diffList = sf.getDiffs().asList();
    SnapshotDiffSection.DiffEntry entry = SnapshotDiffSection.DiffEntry
        .newBuilder().setInodeId(dir.getId()).setType(Type.DIRECTORYDIFF)
        .setNumOfDiff(diffList.size()).build();
    entry.writeDelimitedTo(out);
    for (int i = diffList.size() - 1; i >= 0; i--) { // reverse order!
      DirectoryDiff diff = diffList.get(i);
      SnapshotDiffSection.DirectoryDiff.Builder db = SnapshotDiffSection.
          DirectoryDiff.newBuilder().setSnapshotId(diff.getSnapshotId())
                       .setChildrenSize(diff.getChildrenSize())
                       .setIsSnapshotRoot(diff.isSnapshotRoot());
      INodeDirectoryAttributes copy = diff.snapshotINode;
      if (!diff.isSnapshotRoot() && copy != null) {
        db.setName(ByteString.copyFrom(copy.getLocalNameBytes()))
            .setSnapshotCopy(
                buildINodeDirectory(copy, parent.getSaverContext()));
      }
      // process created list and deleted list
      List<INode> created = diff.getChildrenDiff()
          .getList(ListType.CREATED);
      db.setCreatedListSize(created.size());
      List<INode> deleted = diff.getChildrenDiff().getList(ListType.DELETED);
      for (INode d : deleted) {
        if (d.isReference()) {
          refList.add(d.asReference());
          db.addDeletedINodeRef(refList.size() - 1);
        } else {
          db.addDeletedINode(d.getId());
        }
      }
      db.build().writeDelimitedTo(out);
      saveCreatedList(created, out);
    }
  }
}

/**
 * Make sure we clean the whole subtree under a DstReference node after 
 * deleting a snapshot.
 * see HDFS-5476.
 */
@Test
public void testCleanDstReference() throws Exception {
  final Path test = new Path("/test");
  final Path foo = new Path(test, "foo");
  final Path bar = new Path(foo, "bar");
  hdfs.mkdirs(bar);
  SnapshotTestHelper.createSnapshot(hdfs, test, "s0");

  // create file after s0 so that the file should not be included in s0
  final Path fileInBar = new Path(bar, "file");
  DFSTestUtil.createFile(hdfs, fileInBar, BLOCKSIZE, REPL, SEED);
  // rename foo --> foo2
  final Path foo2 = new Path(test, "foo2");
  hdfs.rename(foo, foo2);
  // create snapshot s1, note the file is included in s1
  hdfs.createSnapshot(test, "s1");
  // delete bar and foo2
  hdfs.delete(new Path(foo2, "bar"), true);
  hdfs.delete(foo2, true);

  final Path sfileInBar = SnapshotTestHelper.getSnapshotPath(test, "s1",
      "foo2/bar/file");
  assertTrue(hdfs.exists(sfileInBar));

  hdfs.deleteSnapshot(test, "s1");
  assertFalse(hdfs.exists(sfileInBar));

  restartClusterAndCheckImage(true);
  // make sure the file under bar is deleted 
  final Path barInS0 = SnapshotTestHelper.getSnapshotPath(test, "s0",
      "foo/bar");
  INodeDirectory barNode = fsdir.getINode(barInS0.toString()).asDirectory();
  assertEquals(0, barNode.getChildrenList(Snapshot.CURRENT_STATE_ID).size());
  List<DirectoryDiff> diffList = barNode.getDiffs().asList();
  assertEquals(1, diffList.size());
  DirectoryDiff diff = diffList.get(0);
  assertEquals(0, diff.getChildrenDiff().getList(ListType.DELETED).size());
  assertEquals(0, diff.getChildrenDiff().getList(ListType.CREATED).size());
}

private void serializeDirDiffList(INodeDirectory dir,
    final List<INodeReference> refList, OutputStream out)
    throws IOException {
  DirectoryWithSnapshotFeature sf = dir.getDirectoryWithSnapshotFeature();
  if (sf != null) {
    List<DirectoryDiff> diffList = sf.getDiffs().asList();
    SnapshotDiffSection.DiffEntry entry = SnapshotDiffSection.DiffEntry
        .newBuilder().setInodeId(dir.getId()).setType(Type.DIRECTORYDIFF)
        .setNumOfDiff(diffList.size()).build();
    entry.writeDelimitedTo(out);
    for (int i = diffList.size() - 1; i >= 0; i--) { // reverse order!
      DirectoryDiff diff = diffList.get(i);
      SnapshotDiffSection.DirectoryDiff.Builder db = SnapshotDiffSection.
          DirectoryDiff.newBuilder().setSnapshotId(diff.getSnapshotId())
                       .setChildrenSize(diff.getChildrenSize())
                       .setIsSnapshotRoot(diff.isSnapshotRoot());
      INodeDirectoryAttributes copy = diff.snapshotINode;
      if (!diff.isSnapshotRoot() && copy != null) {
        db.setName(ByteString.copyFrom(copy.getLocalNameBytes()))
            .setSnapshotCopy(
                buildINodeDirectory(copy, parent.getSaverContext()));
      }
      // process created list and deleted list
      List<INode> created = diff.getChildrenDiff()
          .getList(ListType.CREATED);
      db.setCreatedListSize(created.size());
      List<INode> deleted = diff.getChildrenDiff().getList(ListType.DELETED);
      for (INode d : deleted) {
        if (d.isReference()) {
          refList.add(d.asReference());
          db.addDeletedINodeRef(refList.size() - 1);
        } else {
          db.addDeletedINode(d.getId());
        }
      }
      db.build().writeDelimitedTo(out);
      saveCreatedList(created, out);
    }
  }
}

/**
 * Make sure we clean the whole subtree under a DstReference node after 
 * deleting a snapshot.
 * see HDFS-5476.
 */
@Test
public void testCleanDstReference() throws Exception {
  final Path test = new Path("/test");
  final Path foo = new Path(test, "foo");
  final Path bar = new Path(foo, "bar");
  hdfs.mkdirs(bar);
  SnapshotTestHelper.createSnapshot(hdfs, test, "s0");

  // create file after s0 so that the file should not be included in s0
  final Path fileInBar = new Path(bar, "file");
  DFSTestUtil.createFile(hdfs, fileInBar, BLOCKSIZE, REPL, SEED);
  // rename foo --> foo2
  final Path foo2 = new Path(test, "foo2");
  hdfs.rename(foo, foo2);
  // create snapshot s1, note the file is included in s1
  hdfs.createSnapshot(test, "s1");
  // delete bar and foo2
  hdfs.delete(new Path(foo2, "bar"), true);
  hdfs.delete(foo2, true);

  final Path sfileInBar = SnapshotTestHelper.getSnapshotPath(test, "s1",
      "foo2/bar/file");
  assertTrue(hdfs.exists(sfileInBar));

  hdfs.deleteSnapshot(test, "s1");
  assertFalse(hdfs.exists(sfileInBar));

  restartClusterAndCheckImage(true);
  // make sure the file under bar is deleted 
  final Path barInS0 = SnapshotTestHelper.getSnapshotPath(test, "s0",
      "foo/bar");
  INodeDirectory barNode = fsdir.getINode(barInS0.toString()).asDirectory();
  assertEquals(0, barNode.getChildrenList(Snapshot.CURRENT_STATE_ID).size());
  List<DirectoryDiff> diffList = barNode.getDiffs().asList();
  assertEquals(1, diffList.size());
  DirectoryDiff diff = diffList.get(0);
  assertEquals(0, diff.getChildrenDiff().getList(ListType.DELETED).size());
  assertEquals(0, diff.getChildrenDiff().getList(ListType.CREATED).size());
}

private void serializeDirDiffList(INodeDirectory dir,
    final List<INodeReference> refList, OutputStream out)
    throws IOException {
  DirectoryWithSnapshotFeature sf = dir.getDirectoryWithSnapshotFeature();
  if (sf != null) {
    List<DirectoryDiff> diffList = sf.getDiffs().asList();
    SnapshotDiffSection.DiffEntry entry = SnapshotDiffSection.DiffEntry
        .newBuilder().setInodeId(dir.getId()).setType(Type.DIRECTORYDIFF)
        .setNumOfDiff(diffList.size()).build();
    entry.writeDelimitedTo(out);
    for (int i = diffList.size() - 1; i >= 0; i--) { // reverse order!
      DirectoryDiff diff = diffList.get(i);
      SnapshotDiffSection.DirectoryDiff.Builder db = SnapshotDiffSection.
          DirectoryDiff.newBuilder().setSnapshotId(diff.getSnapshotId())
                       .setChildrenSize(diff.getChildrenSize())
                       .setIsSnapshotRoot(diff.isSnapshotRoot());
      INodeDirectoryAttributes copy = diff.snapshotINode;
      if (!diff.isSnapshotRoot() && copy != null) {
        db.setName(ByteString.copyFrom(copy.getLocalNameBytes()))
            .setSnapshotCopy(
                buildINodeDirectory(copy, parent.getSaverContext()));
      }
      // process created list and deleted list
      List<INode> created = diff.getChildrenDiff()
          .getList(ListType.CREATED);
      db.setCreatedListSize(created.size());
      List<INode> deleted = diff.getChildrenDiff().getList(ListType.DELETED);
      for (INode d : deleted) {
        if (d.isReference()) {
          refList.add(d.asReference());
          db.addDeletedINodeRef(refList.size() - 1);
        } else {
          db.addDeletedINode(d.getId());
        }
      }
      db.build().writeDelimitedTo(out);
      saveCreatedList(created, out);
    }
  }
}

/**
 * Make sure we clean the whole subtree under a DstReference node after 
 * deleting a snapshot.
 * see HDFS-5476.
 */
@Test
public void testCleanDstReference() throws Exception {
  final Path test = new Path("/test");
  final Path foo = new Path(test, "foo");
  final Path bar = new Path(foo, "bar");
  hdfs.mkdirs(bar);
  SnapshotTestHelper.createSnapshot(hdfs, test, "s0");

  // create file after s0 so that the file should not be included in s0
  final Path fileInBar = new Path(bar, "file");
  DFSTestUtil.createFile(hdfs, fileInBar, BLOCKSIZE, REPL, SEED);
  // rename foo --> foo2
  final Path foo2 = new Path(test, "foo2");
  hdfs.rename(foo, foo2);
  // create snapshot s1, note the file is included in s1
  hdfs.createSnapshot(test, "s1");
  // delete bar and foo2
  hdfs.delete(new Path(foo2, "bar"), true);
  hdfs.delete(foo2, true);

  final Path sfileInBar = SnapshotTestHelper.getSnapshotPath(test, "s1",
      "foo2/bar/file");
  assertTrue(hdfs.exists(sfileInBar));

  hdfs.deleteSnapshot(test, "s1");
  assertFalse(hdfs.exists(sfileInBar));

  restartClusterAndCheckImage(true);
  // make sure the file under bar is deleted 
  final Path barInS0 = SnapshotTestHelper.getSnapshotPath(test, "s0",
      "foo/bar");
  INodeDirectory barNode = fsdir.getINode(barInS0.toString()).asDirectory();
  assertEquals(0, barNode.getChildrenList(Snapshot.CURRENT_STATE_ID).size());
  List<DirectoryDiff> diffList = barNode.getDiffs().asList();
  assertEquals(1, diffList.size());
  DirectoryDiff diff = diffList.get(0);
  assertEquals(0, diff.getChildrenDiff().getList(ListType.DELETED).size());
  assertEquals(0, diff.getChildrenDiff().getList(ListType.CREATED).size());
}

/** Add a snapshot. */
Snapshot addSnapshot(int id, String name) throws SnapshotException,
    QuotaExceededException {
  //check snapshot quota
  final int n = getNumSnapshots();
  if (n + 1 > snapshotQuota) {
    throw new SnapshotException("Failed to add snapshot: there are already "
        + n + " snapshot(s) and the snapshot quota is "
        + snapshotQuota);
  }
  final Snapshot s = new Snapshot(id, name, this);
  final byte[] nameBytes = s.getRoot().getLocalNameBytes();
  final int i = searchSnapshot(nameBytes);
  if (i >= 0) {
    throw new SnapshotException("Failed to add snapshot: there is already a "
        + "snapshot with the same name \"" + Snapshot.getSnapshotName(s) + "\".");
  }

  final DirectoryDiff d = getDiffs().addDiff(id, this);
  d.setSnapshotRoot(s.getRoot());
  snapshotsByNames.add(-i - 1, s);

  //set modification time
  updateModificationTime(Time.now(), Snapshot.CURRENT_STATE_ID);
  s.getRoot().setModificationTime(getModificationTime(),
      Snapshot.CURRENT_STATE_ID);
  return s;
}

private void serializeDirDiffList(INodeDirectory dir,
    final List<INodeReference> refList, OutputStream out)
    throws IOException {
  DirectoryWithSnapshotFeature sf = dir.getDirectoryWithSnapshotFeature();
  if (sf != null) {
    List<DirectoryDiff> diffList = sf.getDiffs().asList();
    SnapshotDiffSection.DiffEntry entry = SnapshotDiffSection.DiffEntry
        .newBuilder().setInodeId(dir.getId()).setType(Type.DIRECTORYDIFF)
        .setNumOfDiff(diffList.size()).build();
    entry.writeDelimitedTo(out);
    for (int i = diffList.size() - 1; i >= 0; i--) { // reverse order!
      DirectoryDiff diff = diffList.get(i);
      SnapshotDiffSection.DirectoryDiff.Builder db = SnapshotDiffSection.
          DirectoryDiff.newBuilder().setSnapshotId(diff.getSnapshotId())
                       .setChildrenSize(diff.getChildrenSize())
                       .setIsSnapshotRoot(diff.isSnapshotRoot());
      INodeDirectoryAttributes copy = diff.snapshotINode;
      if (!diff.isSnapshotRoot() && copy != null) {
        db.setName(ByteString.copyFrom(copy.getLocalNameBytes()))
            .setSnapshotCopy(
                buildINodeDirectory(copy, parent.getSaverContext()));
      }
      // process created list and deleted list
      List<INode> created = diff.getChildrenDiff()
          .getList(ListType.CREATED);
      db.setCreatedListSize(created.size());
      List<INode> deleted = diff.getChildrenDiff().getList(ListType.DELETED);
      for (INode d : deleted) {
        if (d.isReference()) {
          refList.add(d.asReference());
          db.addDeletedINodeRef(refList.size() - 1);
        } else {
          db.addDeletedINode(d.getId());
        }
      }
      db.build().writeDelimitedTo(out);
      saveCreatedList(created, out);
    }
  }
}

/**
 * Check the correctness of snapshot list within
 * {@link INodeDirectorySnapshottable}
 */
private void checkSnapshotList(INodeDirectorySnapshottable srcRoot,
    String[] sortedNames, String[] names) {
  ReadOnlyList<Snapshot> listByName = srcRoot.getSnapshotsByNames();
  assertEquals(sortedNames.length, listByName.size());
  for (int i = 0; i < listByName.size(); i++) {
    assertEquals(sortedNames[i], listByName.get(i).getRoot().getLocalName());
  }
  List<DirectoryDiff> listByTime = srcRoot.getDiffs().asList();
  assertEquals(names.length, listByTime.size());
  for (int i = 0; i < listByTime.size(); i++) {
    Snapshot s = srcRoot.getSnapshotById(listByTime.get(i).getSnapshotId());
    assertEquals(names[i], s.getRoot().getLocalName());
  }
}

/**
 * Make sure we clean the whole subtree under a DstReference node after 
 * deleting a snapshot.
 * see HDFS-5476.
 */
@Test
public void testCleanDstReference() throws Exception {
  final Path test = new Path("/test");
  final Path foo = new Path(test, "foo");
  final Path bar = new Path(foo, "bar");
  hdfs.mkdirs(bar);
  SnapshotTestHelper.createSnapshot(hdfs, test, "s0");

  // create file after s0 so that the file should not be included in s0
  final Path fileInBar = new Path(bar, "file");
  DFSTestUtil.createFile(hdfs, fileInBar, BLOCKSIZE, REPL, SEED);
  // rename foo --> foo2
  final Path foo2 = new Path(test, "foo2");
  hdfs.rename(foo, foo2);
  // create snapshot s1, note the file is included in s1
  hdfs.createSnapshot(test, "s1");
  // delete bar and foo2
  hdfs.delete(new Path(foo2, "bar"), true);
  hdfs.delete(foo2, true);

  final Path sfileInBar = SnapshotTestHelper.getSnapshotPath(test, "s1",
      "foo2/bar/file");
  assertTrue(hdfs.exists(sfileInBar));

  hdfs.deleteSnapshot(test, "s1");
  assertFalse(hdfs.exists(sfileInBar));

  restartClusterAndCheckImage(true);
  // make sure the file under bar is deleted 
  final Path barInS0 = SnapshotTestHelper.getSnapshotPath(test, "s0",
      "foo/bar");
  INodeDirectory barNode = fsdir.getINode(barInS0.toString()).asDirectory();
  assertEquals(0, barNode.getChildrenList(Snapshot.CURRENT_STATE_ID).size());
  List<DirectoryDiff> diffList = barNode.getDiffs().asList();
  assertEquals(1, diffList.size());
  DirectoryDiff diff = diffList.get(0);
  assertEquals(0, diff.getChildrenDiff().getList(ListType.DELETED).size());
  assertEquals(0, diff.getChildrenDiff().getList(ListType.CREATED).size());
}

/**
 * Test when there is snapshot taken on root
 */
@Test
public void testSnapshotOnRoot() throws Exception {
  final Path root = new Path("/");
  hdfs.allowSnapshot(root);
  hdfs.createSnapshot(root, "s1");

  cluster.shutdown();
  cluster = new MiniDFSCluster.Builder(conf).format(false)
      .numDataNodes(REPLICATION).build();
  cluster.waitActive();
  fsn = cluster.getNamesystem();
  hdfs = cluster.getFileSystem();

  // save namespace and restart cluster
  hdfs.setSafeMode(SafeModeAction.SAFEMODE_ENTER);
  hdfs.saveNamespace();
  hdfs.setSafeMode(SafeModeAction.SAFEMODE_LEAVE);
  cluster.shutdown();
  cluster = new MiniDFSCluster.Builder(conf).format(false)
      .numDataNodes(REPLICATION).build();
  cluster.waitActive();
  fsn = cluster.getNamesystem();
  hdfs = cluster.getFileSystem();

  INodeDirectory rootNode = fsn.dir.getINode4Write(root.toString())
      .asDirectory();
  assertTrue("The children list of root should be empty", 
      rootNode.getChildrenList(Snapshot.CURRENT_STATE_ID).isEmpty());
  // one snapshot on root: s1
  List<DirectoryDiff> diffList = rootNode.getDiffs().asList();
  assertEquals(1, diffList.size());
  Snapshot s1 = rootNode.getSnapshot(DFSUtil.string2Bytes("s1"));
  assertEquals(s1.getId(), diffList.get(0).getSnapshotId());

  // check SnapshotManager's snapshottable directory list
  assertEquals(1, fsn.getSnapshotManager().getNumSnapshottableDirs());
  SnapshottableDirectoryStatus[] sdirs = fsn.getSnapshotManager()
      .getSnapshottableDirListing(null);
  assertEquals(root, sdirs[0].getFullPath());

  // save namespace and restart cluster
  hdfs.setSafeMode(SafeModeAction.SAFEMODE_ENTER);
  hdfs.saveNamespace();
  hdfs.setSafeMode(SafeModeAction.SAFEMODE_LEAVE);
  cluster.shutdown();
  cluster = new MiniDFSCluster.Builder(conf).format(false)
      .numDataNodes(REPLICATION).build();
  cluster.waitActive();
  fsn = cluster.getNamesystem();
  hdfs = cluster.getFileSystem();
}

/**
 * Test the undo section of rename. Before the rename, we create the renamed 
 * file/dir before taking the snapshot.
 */
@Test
public void testRenameUndo_1() throws Exception {
  final Path sdir1 = new Path("/dir1");
  final Path sdir2 = new Path("/dir2");
  hdfs.mkdirs(sdir1);
  hdfs.mkdirs(sdir2);
  final Path foo = new Path(sdir1, "foo");
  final Path bar = new Path(foo, "bar");
  DFSTestUtil.createFile(hdfs, bar, BLOCKSIZE, REPL, SEED);
  final Path dir2file = new Path(sdir2, "file");
  DFSTestUtil.createFile(hdfs, dir2file, BLOCKSIZE, REPL, SEED);

  SnapshotTestHelper.createSnapshot(hdfs, sdir1, "s1");

  INodeDirectory dir2 = fsdir.getINode4Write(sdir2.toString()).asDirectory();
  INodeDirectory mockDir2 = spy(dir2);
  doReturn(false).when(mockDir2).addChild((INode) anyObject(), anyBoolean(),
         Mockito.anyInt());
  INodeDirectory root = fsdir.getINode4Write("/").asDirectory();
  root.replaceChild(dir2, mockDir2, fsdir.getINodeMap());

  final Path newfoo = new Path(sdir2, "foo");
  boolean result = hdfs.rename(foo, newfoo);
  assertFalse(result);

  // check the current internal details
  INodeDirectory dir1Node = fsdir.getINode4Write(sdir1.toString())
      .asDirectory();
  Snapshot s1 = dir1Node.getSnapshot(DFSUtil.string2Bytes("s1"));
  ReadOnlyList<INode> dir1Children = dir1Node
      .getChildrenList(Snapshot.CURRENT_STATE_ID);
  assertEquals(1, dir1Children.size());
  assertEquals(foo.getName(), dir1Children.get(0).getLocalName());
  List<DirectoryDiff> dir1Diffs = dir1Node.getDiffs().asList();
  assertEquals(1, dir1Diffs.size());
  assertEquals(s1.getId(), dir1Diffs.get(0).getSnapshotId());

  // after the undo of rename, both the created and deleted list of sdir1
  // should be empty
  ChildrenDiff childrenDiff = dir1Diffs.get(0).getChildrenDiff();
  assertEquals(0, childrenDiff.getList(ListType.DELETED).size());
  assertEquals(0, childrenDiff.getList(ListType.CREATED).size());

  INode fooNode = fsdir.getINode4Write(foo.toString());
  assertTrue(fooNode.isDirectory() && fooNode.asDirectory().isWithSnapshot());
  List<DirectoryDiff> fooDiffs = fooNode.asDirectory().getDiffs().asList();
  assertEquals(1, fooDiffs.size());
  assertEquals(s1.getId(), fooDiffs.get(0).getSnapshotId());

  final Path foo_s1 = SnapshotTestHelper.getSnapshotPath(sdir1, "s1", "foo");
  INode fooNode_s1 = fsdir.getINode(foo_s1.toString());
  assertTrue(fooNode_s1 == fooNode);

  // check sdir2
  assertFalse(hdfs.exists(newfoo));
  INodeDirectory dir2Node = fsdir.getINode4Write(sdir2.toString())
      .asDirectory();
  assertFalse(dir2Node.isWithSnapshot());
  ReadOnlyList<INode> dir2Children = dir2Node
      .getChildrenList(Snapshot.CURRENT_STATE_ID);
  assertEquals(1, dir2Children.size());
  assertEquals(dir2file.getName(), dir2Children.get(0).getLocalName());
}

/**
 * Test the undo section of rename. Before the rename, we create the renamed 
 * file/dir after taking the snapshot.
 */
@Test
public void testRenameUndo_2() throws Exception {
  final Path sdir1 = new Path("/dir1");
  final Path sdir2 = new Path("/dir2");
  hdfs.mkdirs(sdir1);
  hdfs.mkdirs(sdir2);
  final Path dir2file = new Path(sdir2, "file");
  DFSTestUtil.createFile(hdfs, dir2file, BLOCKSIZE, REPL, SEED);

  SnapshotTestHelper.createSnapshot(hdfs, sdir1, "s1");

  // create foo after taking snapshot
  final Path foo = new Path(sdir1, "foo");
  final Path bar = new Path(foo, "bar");
  DFSTestUtil.createFile(hdfs, bar, BLOCKSIZE, REPL, SEED);

  INodeDirectory dir2 = fsdir.getINode4Write(sdir2.toString()).asDirectory();
  INodeDirectory mockDir2 = spy(dir2);
  doReturn(false).when(mockDir2).addChild((INode) anyObject(), anyBoolean(),
          Mockito.anyInt());
  INodeDirectory root = fsdir.getINode4Write("/").asDirectory();
  root.replaceChild(dir2, mockDir2, fsdir.getINodeMap());

  final Path newfoo = new Path(sdir2, "foo");
  boolean result = hdfs.rename(foo, newfoo);
  assertFalse(result);

  // check the current internal details
  INodeDirectory dir1Node = fsdir.getINode4Write(sdir1.toString())
      .asDirectory();
  Snapshot s1 = dir1Node.getSnapshot(DFSUtil.string2Bytes("s1"));
  ReadOnlyList<INode> dir1Children = dir1Node
      .getChildrenList(Snapshot.CURRENT_STATE_ID);
  assertEquals(1, dir1Children.size());
  assertEquals(foo.getName(), dir1Children.get(0).getLocalName());
  List<DirectoryDiff> dir1Diffs = dir1Node.getDiffs().asList();
  assertEquals(1, dir1Diffs.size());
  assertEquals(s1.getId(), dir1Diffs.get(0).getSnapshotId());

  // after the undo of rename, the created list of sdir1 should contain 
  // 1 element
  ChildrenDiff childrenDiff = dir1Diffs.get(0).getChildrenDiff();
  assertEquals(0, childrenDiff.getList(ListType.DELETED).size());
  assertEquals(1, childrenDiff.getList(ListType.CREATED).size());

  INode fooNode = fsdir.getINode4Write(foo.toString());
  assertTrue(fooNode instanceof INodeDirectory);
  assertTrue(childrenDiff.getList(ListType.CREATED).get(0) == fooNode);

  final Path foo_s1 = SnapshotTestHelper.getSnapshotPath(sdir1, "s1", "foo");
  assertFalse(hdfs.exists(foo_s1));

  // check sdir2
  assertFalse(hdfs.exists(newfoo));
  INodeDirectory dir2Node = fsdir.getINode4Write(sdir2.toString())
      .asDirectory();
  assertFalse(dir2Node.isWithSnapshot());
  ReadOnlyList<INode> dir2Children = dir2Node
      .getChildrenList(Snapshot.CURRENT_STATE_ID);
  assertEquals(1, dir2Children.size());
  assertEquals(dir2file.getName(), dir2Children.get(0).getLocalName());
}

/**
 * Test rename while the rename operation will exceed the quota in the dst
 * tree.
 */
@Test
public void testRenameUndo_5() throws Exception {
  final Path test = new Path("/test");
  final Path dir1 = new Path(test, "dir1");
  final Path dir2 = new Path(test, "dir2");
  final Path subdir2 = new Path(dir2, "subdir2");
  hdfs.mkdirs(dir1);
  hdfs.mkdirs(subdir2);

  final Path foo = new Path(dir1, "foo");
  final Path bar = new Path(foo, "bar");
  DFSTestUtil.createFile(hdfs, bar, BLOCKSIZE, REPL, SEED);

  SnapshotTestHelper.createSnapshot(hdfs, dir1, "s1");
  SnapshotTestHelper.createSnapshot(hdfs, dir2, "s2");

  // set ns quota of dir2 to 4, so the current remaining is 2 (already has
  // dir2, and subdir2)
  hdfs.setQuota(dir2, 4, Long.MAX_VALUE - 1);

  final Path foo2 = new Path(subdir2, foo.getName());
  FSDirectory fsdir2 = Mockito.spy(fsdir);
  Mockito.doThrow(new NSQuotaExceededException("fake exception")).when(fsdir2)
      .addLastINode((INodesInPath) Mockito.anyObject(),
          (INode) Mockito.anyObject(), Mockito.anyBoolean());
  Whitebox.setInternalState(fsn, "dir", fsdir2);
  // rename /test/dir1/foo to /test/dir2/subdir2/foo. 
  // FSDirectory#verifyQuota4Rename will pass since the remaining quota is 2.
  // However, the rename operation will fail since we let addLastINode throw
  // NSQuotaExceededException
  boolean rename = hdfs.rename(foo, foo2);
  assertFalse(rename);

  // check the undo
  assertTrue(hdfs.exists(foo));
  assertTrue(hdfs.exists(bar));
  INodeDirectory dir1Node = fsdir2.getINode4Write(dir1.toString())
      .asDirectory();
  List<INode> childrenList = ReadOnlyList.Util.asList(dir1Node
      .getChildrenList(Snapshot.CURRENT_STATE_ID));
  assertEquals(1, childrenList.size());
  INode fooNode = childrenList.get(0);
  assertTrue(fooNode.asDirectory().isWithSnapshot());
  INode barNode = fsdir2.getINode4Write(bar.toString());
  assertTrue(barNode.getClass() == INodeFile.class);
  assertSame(fooNode, barNode.getParent());
  List<DirectoryDiff> diffList = dir1Node
      .getDiffs().asList();
  assertEquals(1, diffList.size());
  DirectoryDiff diff = diffList.get(0);
  assertTrue(diff.getChildrenDiff().getList(ListType.CREATED).isEmpty());
  assertTrue(diff.getChildrenDiff().getList(ListType.DELETED).isEmpty());

  // check dir2
  INodeDirectory dir2Node = fsdir2.getINode4Write(dir2.toString()).asDirectory();
  assertTrue(dir2Node.isSnapshottable());
  QuotaCounts counts = dir2Node.computeQuotaUsage(fsdir.getBlockStoragePolicySuite());
  assertEquals(2, counts.getNameSpace());
  assertEquals(0, counts.getStorageSpace());
  childrenList = ReadOnlyList.Util.asList(dir2Node.asDirectory()
      .getChildrenList(Snapshot.CURRENT_STATE_ID));
  assertEquals(1, childrenList.size());
  INode subdir2Node = childrenList.get(0);
  assertSame(dir2Node, subdir2Node.getParent());
  assertSame(subdir2Node, fsdir2.getINode4Write(subdir2.toString()));
  diffList = dir2Node.getDiffs().asList();
  assertEquals(1, diffList.size());
  diff = diffList.get(0);
  assertTrue(diff.getChildrenDiff().getList(ListType.CREATED).isEmpty());
  assertTrue(diff.getChildrenDiff().getList(ListType.DELETED).isEmpty());
}