/**
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.apache.hadoop.hdfs.server.namenode.snapshot;

import java.io.IOException;
import java.util.Arrays;
import java.util.EnumSet;
import java.util.HashSet;
import java.util.Random;
import java.util.Set;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicBoolean;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileChecksum;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hdfs.DFSConfigKeys;
import org.apache.hadoop.hdfs.DFSOutputStream;
import org.apache.hadoop.hdfs.DFSTestUtil;
import org.apache.hadoop.hdfs.DistributedFileSystem;
import org.apache.hadoop.hdfs.MiniDFSCluster;
import org.apache.hadoop.hdfs.client.HdfsDataOutputStream;
import org.apache.hadoop.hdfs.client.HdfsDataOutputStream.SyncFlag;
import org.apache.hadoop.hdfs.protocol.HdfsConstants;
import org.apache.hadoop.hdfs.server.namenode.NameNode;
import org.apache.hadoop.hdfs.server.namenode.NameNodeAdapter;
import org.apache.hadoop.hdfs.server.protocol.NamenodeProtocols;
import org.apache.hadoop.util.Time;
import org.junit.After;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;

public class TestOpenFilesWithSnapshot {
  private static final Logger LOG =
      LoggerFactory.getLogger(TestOpenFilesWithSnapshot.class.getName());
  private final Configuration conf = new Configuration();
  MiniDFSCluster cluster = null;
  DistributedFileSystem fs = null;

  private static final long SEED = 0;
  private static final short REPLICATION = 3;
  private static final long BLOCKSIZE = 1024;
  private static final long BUFFERLEN = BLOCKSIZE / 2;
  private static final long FILELEN = BLOCKSIZE * 2;

  @Before
  public void setup() throws IOException {
    conf.setBoolean(
        DFSConfigKeys.DFS_NAMENODE_SNAPSHOT_CAPTURE_OPENFILES, true);
    cluster =
        new MiniDFSCluster.Builder(conf).numDataNodes(REPLICATION).build();
    conf.set("dfs.blocksize", "1048576");
    fs = cluster.getFileSystem();
  }

  @After
  public void teardown() throws IOException {
    if (fs != null) {
      fs.close();
      fs = null;
    }
    if (cluster != null) {
      cluster.shutdown();
      cluster = null;
    }

  }

  @Test
  public void testUCFileDeleteWithSnapShot() throws Exception {
    Path path = new Path("/test");
    doWriteAndAbort(fs, path);

    // delete files separately
    fs.delete(new Path("/test/test/test2"), true);
    fs.delete(new Path("/test/test/test3"), true);
    restartNameNode();
  }

  @Test
  public void testParentDirWithUCFileDeleteWithSnapShot() throws Exception {
    Path path = new Path("/test");
    doWriteAndAbort(fs, path);

    // delete parent directory
    fs.delete(new Path("/test/test"), true);
    restartNameNode();
  }

  @Test
  public void testWithCheckpoint() throws Exception {
    Path path = new Path("/test");
    doWriteAndAbort(fs, path);
    fs.delete(new Path("/test/test"), true);
    restartNameNode();
    
    // read snapshot file after restart
    String test2snapshotPath = Snapshot.getSnapshotPath(path.toString(),
        "s1/test/test2");
    DFSTestUtil.readFile(fs, new Path(test2snapshotPath));
    String test3snapshotPath = Snapshot.getSnapshotPath(path.toString(),
        "s1/test/test3");
    DFSTestUtil.readFile(fs, new Path(test3snapshotPath));
  }

  @Test
  public void testFilesDeletionWithCheckpoint() throws Exception {
    Path path = new Path("/test");
    doWriteAndAbort(fs, path);
    fs.delete(new Path("/test/test/test2"), true);
    fs.delete(new Path("/test/test/test3"), true);
    restartNameNode();
    
    // read snapshot file after restart
    String test2snapshotPath = Snapshot.getSnapshotPath(path.toString(),
        "s1/test/test2");
    DFSTestUtil.readFile(fs, new Path(test2snapshotPath));
    String test3snapshotPath = Snapshot.getSnapshotPath(path.toString(),
        "s1/test/test3");
    DFSTestUtil.readFile(fs, new Path(test3snapshotPath));
  }

  private void doWriteAndAbort(DistributedFileSystem fs, Path path)
      throws IOException {
    fs.mkdirs(path);
    fs.allowSnapshot(path);
    DFSTestUtil
        .createFile(fs, new Path("/test/test1"), 100, (short) 2, 100024L);
    DFSTestUtil
        .createFile(fs, new Path("/test/test2"), 100, (short) 2, 100024L);
    Path file = new Path("/test/test/test2");
    FSDataOutputStream out = fs.create(file);
    for (int i = 0; i < 2; i++) {
      long count = 0;
      while (count < 1048576) {
        out.writeBytes("hell");
        count += 4;
      }
    }
    ((DFSOutputStream) out.getWrappedStream()).hsync(EnumSet
        .of(SyncFlag.UPDATE_LENGTH));
    DFSTestUtil.abortStream((DFSOutputStream) out.getWrappedStream());
    Path file2 = new Path("/test/test/test3");
    FSDataOutputStream out2 = fs.create(file2);
    for (int i = 0; i < 2; i++) {
      long count = 0;
      while (count < 1048576) {
        out2.writeBytes("hell");
        count += 4;
      }
    }
    ((DFSOutputStream) out2.getWrappedStream()).hsync(EnumSet
        .of(SyncFlag.UPDATE_LENGTH));
    DFSTestUtil.abortStream((DFSOutputStream) out2.getWrappedStream());
    fs.createSnapshot(path, "s1");
  }

  @Test
  public void testOpenFilesWithMultipleSnapshots() throws Exception {
    doTestMultipleSnapshots(true);
  }

  @Test
  public void testOpenFilesWithMultipleSnapshotsWithoutCheckpoint()
      throws Exception {
    doTestMultipleSnapshots(false);
  }

  private void doTestMultipleSnapshots(boolean saveNamespace)
      throws IOException {
    Path path = new Path("/test");
    doWriteAndAbort(fs, path);
    fs.createSnapshot(path, "s2");
    fs.delete(new Path("/test/test"), true);
    fs.deleteSnapshot(path, "s2");
    cluster.triggerBlockReports();
    if (saveNamespace) {
      NameNode nameNode = cluster.getNameNode();
      NameNodeAdapter.enterSafeMode(nameNode, false);
      NameNodeAdapter.saveNamespace(nameNode);
      NameNodeAdapter.leaveSafeMode(nameNode);
    }
    cluster.restartNameNode(true);
  }
  
  @Test
  public void testOpenFilesWithRename() throws Exception {
    Path path = new Path("/test");
    doWriteAndAbort(fs, path);

    // check for zero sized blocks
    Path fileWithEmptyBlock = new Path("/test/test/test4");
    fs.create(fileWithEmptyBlock);
    NamenodeProtocols nameNodeRpc = cluster.getNameNodeRpc();
    String clientName = fs.getClient().getClientName();
    // create one empty block
    nameNodeRpc.addBlock(fileWithEmptyBlock.toString(), clientName, null, null,
        HdfsConstants.GRANDFATHER_INODE_ID, null, null);
    fs.createSnapshot(path, "s2");

    fs.rename(new Path("/test/test"), new Path("/test/test-renamed"));
    fs.delete(new Path("/test/test-renamed"), true);
    restartNameNode();
  }

  private void createFile(final Path filePath) throws IOException {
    DFSTestUtil.createFile(fs, filePath, (int) BUFFERLEN,
        FILELEN, BLOCKSIZE, REPLICATION, SEED);
  }

  private int writeToStream(final FSDataOutputStream outputStream, byte[] buf)
      throws IOException {
    outputStream.write(buf);
    ((HdfsDataOutputStream)outputStream).hsync(
        EnumSet.of(SyncFlag.UPDATE_LENGTH));
    return buf.length;
  }

  /**
   * Test open files under snapshot directories are getting captured
   * in snapshots as a truly immutable copy. Verify open files outside
   * of snapshot directory not getting affected.
   *
   * \- level_0_A
   *   \- level_1_C
   *     +- appA.log         (open file, not under snap root)
   *     \- level_2_E        (Snapshottable Dir)
   *       \- level_3_G
   *         +- flume.log    (open file, under snap root)
   * \- level_0_B
   *   +- appB.log         (open file, not under snap root)
   *   \- level_2_D        (Snapshottable Dir)
   *     +- hbase.log      (open file, under snap root)
   */
  @Test (timeout = 120000)
  public void testPointInTimeSnapshotCopiesForOpenFiles() throws Exception {
    // Construct the directory tree
    final Path level0A = new Path("/level_0_A");
    final Path level0B = new Path("/level_0_B");
    final Path level1C = new Path(level0A, "level_1_C");
    final Path level1D = new Path(level0B, "level_1_D");
    final Path level2E = new Path(level1C, "level_2_E");
    final Path level3G = new Path(level2E, "level_3_G");
    Set<Path> dirPaths = new HashSet<>(Arrays.asList(level0A, level0B,
        level1C, level1D, level2E, level3G));
    for (Path dirPath : dirPaths) {
      fs.mkdirs(dirPath);
    }

    // String constants
    final Path flumeSnapRootDir = level2E;
    final Path hbaseSnapRootDir = level1D;
    final String flumeFileName = "flume.log";
    final String hbaseFileName = "hbase.log";
    final String appAFileName = "appA.log";
    final String appBFileName = "appB.log";
    final String flumeSnap1Name = "flume_snap_s1";
    final String flumeSnap2Name = "flume_snap_s2";
    final String flumeSnap3Name = "flume_snap_s3";
    final String hbaseSnap1Name = "hbase_snap_s1";
    final String hbaseSnap2Name = "hbase_snap_s2";
    final String hbaseSnap3Name = "hbase_snap_s3";
    final String flumeRelPathFromSnapDir = "level_3_G/" + flumeFileName;

    // Create files and open a stream
    final Path flumeFile = new Path(level3G, flumeFileName);
    createFile(flumeFile);
    FSDataOutputStream flumeOutputStream = fs.append(flumeFile);

    final Path hbaseFile = new Path(level1D, hbaseFileName);
    createFile(hbaseFile);
    FSDataOutputStream hbaseOutputStream = fs.append(hbaseFile);

    final Path appAFile = new Path(level1C, appAFileName);
    createFile(appAFile);
    FSDataOutputStream appAOutputStream = fs.append(appAFile);

    final Path appBFile = new Path(level0B, appBFileName);
    createFile(appBFile);
    FSDataOutputStream appBOutputStream = fs.append(appBFile);

    final long appAFileInitialLength = fs.getFileStatus(appAFile).getLen();
    final long appBFileInitialLength = fs.getFileStatus(appBFile).getLen();

    // Create Snapshot S1
    final Path flumeS1Dir = SnapshotTestHelper.createSnapshot(
        fs, flumeSnapRootDir, flumeSnap1Name);
    final Path flumeS1Path = new Path(flumeS1Dir, flumeRelPathFromSnapDir);
    final Path hbaseS1Dir = SnapshotTestHelper.createSnapshot(
        fs, hbaseSnapRootDir, hbaseSnap1Name);
    final Path hbaseS1Path = new Path(hbaseS1Dir, hbaseFileName);

    final long flumeFileLengthAfterS1 = fs.getFileStatus(flumeFile).getLen();
    final long hbaseFileLengthAfterS1 = fs.getFileStatus(hbaseFile).getLen();

    // Verify if Snap S1 file lengths are same as the the live ones
    Assert.assertEquals(flumeFileLengthAfterS1,
        fs.getFileStatus(flumeS1Path).getLen());
    Assert.assertEquals(hbaseFileLengthAfterS1,
        fs.getFileStatus(hbaseS1Path).getLen());
    Assert.assertEquals(appAFileInitialLength,
        fs.getFileStatus(appAFile).getLen());
    Assert.assertEquals(appBFileInitialLength,
        fs.getFileStatus(appBFile).getLen());

    long flumeFileWrittenDataLength = flumeFileLengthAfterS1;
    long hbaseFileWrittenDataLength = hbaseFileLengthAfterS1;
    long appAFileWrittenDataLength = appAFileInitialLength;

    int newWriteLength = (int) (BLOCKSIZE * 1.5);
    byte[] buf = new byte[newWriteLength];
    Random random = new Random();
    random.nextBytes(buf);

    // Write more data to flume and hbase files only
    flumeFileWrittenDataLength += writeToStream(flumeOutputStream, buf);
    hbaseFileWrittenDataLength += writeToStream(hbaseOutputStream, buf);

    // Create Snapshot S2
    final Path flumeS2Dir = SnapshotTestHelper.createSnapshot(
        fs, flumeSnapRootDir, flumeSnap2Name);
    final Path flumeS2Path = new Path(flumeS2Dir, flumeRelPathFromSnapDir);
    final Path hbaseS2Dir = SnapshotTestHelper.createSnapshot(
        fs, hbaseSnapRootDir, hbaseSnap2Name);
    final Path hbaseS2Path = new Path(hbaseS2Dir, hbaseFileName);

    // Verify live files lengths are same as all data written till now
    final long flumeFileLengthAfterS2 = fs.getFileStatus(flumeFile).getLen();
    final long hbaseFileLengthAfterS2 = fs.getFileStatus(hbaseFile).getLen();
    Assert.assertEquals(flumeFileWrittenDataLength, flumeFileLengthAfterS2);
    Assert.assertEquals(hbaseFileWrittenDataLength, hbaseFileLengthAfterS2);

    // Verify if Snap S2 file lengths are same as the live ones
    Assert.assertEquals(flumeFileLengthAfterS2,
        fs.getFileStatus(flumeS2Path).getLen());
    Assert.assertEquals(hbaseFileLengthAfterS2,
        fs.getFileStatus(hbaseS2Path).getLen());
    Assert.assertEquals(appAFileInitialLength,
        fs.getFileStatus(appAFile).getLen());
    Assert.assertEquals(appBFileInitialLength,
        fs.getFileStatus(appBFile).getLen());

    // Write more data to appA file only
    newWriteLength = (int) (BLOCKSIZE * 2.5);
    buf = new byte[newWriteLength];
    random.nextBytes(buf);
    appAFileWrittenDataLength += writeToStream(appAOutputStream, buf);

    // Verify other open files are not affected in their snapshots
    Assert.assertEquals(flumeFileLengthAfterS2,
        fs.getFileStatus(flumeS2Path).getLen());
    Assert.assertEquals(appAFileWrittenDataLength,
        fs.getFileStatus(appAFile).getLen());

    // Write more data to flume file only
    newWriteLength = (int) (BLOCKSIZE * 2.5);
    buf = new byte[newWriteLength];
    random.nextBytes(buf);
    flumeFileWrittenDataLength += writeToStream(flumeOutputStream, buf);

    // Create Snapshot S3
    final Path flumeS3Dir = SnapshotTestHelper.createSnapshot(
        fs, flumeSnapRootDir, flumeSnap3Name);
    final Path flumeS3Path = new Path(flumeS3Dir, flumeRelPathFromSnapDir);
    final Path hbaseS3Dir = SnapshotTestHelper.createSnapshot(
        fs, hbaseSnapRootDir, hbaseSnap3Name);
    final Path hbaseS3Path = new Path(hbaseS3Dir, hbaseFileName);

    // Verify live files lengths are same as all data written till now
    final long flumeFileLengthAfterS3 = fs.getFileStatus(flumeFile).getLen();
    final long hbaseFileLengthAfterS3 = fs.getFileStatus(hbaseFile).getLen();
    Assert.assertEquals(flumeFileWrittenDataLength, flumeFileLengthAfterS3);
    Assert.assertEquals(hbaseFileWrittenDataLength, hbaseFileLengthAfterS3);

    // Verify if Snap S3 file lengths are same as the live ones
    Assert.assertEquals(flumeFileLengthAfterS3,
        fs.getFileStatus(flumeS3Path).getLen());
    Assert.assertEquals(hbaseFileLengthAfterS3,
        fs.getFileStatus(hbaseS3Path).getLen());
    Assert.assertEquals(appAFileWrittenDataLength,
        fs.getFileStatus(appAFile).getLen());
    Assert.assertEquals(appBFileInitialLength,
        fs.getFileStatus(appBFile).getLen());

    // Verify old flume snapshots have point-in-time / frozen file lengths
    // even after the live file have moved forward.
    Assert.assertEquals(flumeFileLengthAfterS1,
        fs.getFileStatus(flumeS1Path).getLen());
    Assert.assertEquals(flumeFileLengthAfterS2,
        fs.getFileStatus(flumeS2Path).getLen());
    Assert.assertEquals(flumeFileLengthAfterS3,
        fs.getFileStatus(flumeS3Path).getLen());

    // Verify old hbase snapshots have point-in-time / frozen file lengths
    // even after the live files have moved forward.
    Assert.assertEquals(hbaseFileLengthAfterS1,
        fs.getFileStatus(hbaseS1Path).getLen());
    Assert.assertEquals(hbaseFileLengthAfterS2,
        fs.getFileStatus(hbaseS2Path).getLen());
    Assert.assertEquals(hbaseFileLengthAfterS3,
        fs.getFileStatus(hbaseS3Path).getLen());

    flumeOutputStream.close();
    hbaseOutputStream.close();
    appAOutputStream.close();
    appBOutputStream.close();
  }

  /**
   * Test snapshot capturing open files and verify the same
   * across NameNode restarts.
   */
  @Test (timeout = 120000)
  public void testSnapshotsForOpenFilesWithNNRestart() throws Exception {
    // Construct the directory tree
    final Path level0A = new Path("/level_0_A");
    final Path flumeSnapRootDir = level0A;
    final String flumeFileName = "flume.log";
    final String flumeSnap1Name = "flume_snap_1";
    final String flumeSnap2Name = "flume_snap_2";

    // Create files and open a stream
    final Path flumeFile = new Path(level0A, flumeFileName);
    createFile(flumeFile);
    FSDataOutputStream flumeOutputStream = fs.append(flumeFile);

    // Create Snapshot S1
    final Path flumeS1Dir = SnapshotTestHelper.createSnapshot(
        fs, flumeSnapRootDir, flumeSnap1Name);
    final Path flumeS1Path = new Path(flumeS1Dir, flumeFileName);
    final long flumeFileLengthAfterS1 = fs.getFileStatus(flumeFile).getLen();

    // Verify if Snap S1 file length is same as the the live one
    Assert.assertEquals(flumeFileLengthAfterS1,
        fs.getFileStatus(flumeS1Path).getLen());

    long flumeFileWrittenDataLength = flumeFileLengthAfterS1;
    int newWriteLength = (int) (BLOCKSIZE * 1.5);
    byte[] buf = new byte[newWriteLength];
    Random random = new Random();
    random.nextBytes(buf);

    // Write more data to flume file
    flumeFileWrittenDataLength += writeToStream(flumeOutputStream, buf);

    // Create Snapshot S2
    final Path flumeS2Dir = SnapshotTestHelper.createSnapshot(
        fs, flumeSnapRootDir, flumeSnap2Name);
    final Path flumeS2Path = new Path(flumeS2Dir, flumeFileName);

    // Verify live files length is same as all data written till now
    final long flumeFileLengthAfterS2 = fs.getFileStatus(flumeFile).getLen();
    Assert.assertEquals(flumeFileWrittenDataLength, flumeFileLengthAfterS2);

    // Verify if Snap S2 file length is same as the live one
    Assert.assertEquals(flumeFileLengthAfterS2,
        fs.getFileStatus(flumeS2Path).getLen());

    // Write more data to flume file
    flumeFileWrittenDataLength += writeToStream(flumeOutputStream, buf);

    // Verify old flume snapshots have point-in-time / frozen file lengths
    // even after the live file have moved forward.
    Assert.assertEquals(flumeFileLengthAfterS1,
        fs.getFileStatus(flumeS1Path).getLen());
    Assert.assertEquals(flumeFileLengthAfterS2,
        fs.getFileStatus(flumeS2Path).getLen());

    // Restart the NameNode
    restartNameNode();
    cluster.waitActive();

    // Verify live file length hasn't changed after NN restart
    Assert.assertEquals(flumeFileWrittenDataLength,
        fs.getFileStatus(flumeFile).getLen());

    // Verify old flume snapshots have point-in-time / frozen file lengths
    // after NN restart and live file moved forward.
    Assert.assertEquals(flumeFileLengthAfterS1,
        fs.getFileStatus(flumeS1Path).getLen());
    Assert.assertEquals(flumeFileLengthAfterS2,
        fs.getFileStatus(flumeS2Path).getLen());

    flumeOutputStream.close();
  }

  /**
   * Test snapshot capturing open files when an open file with active lease
   * is deleted by the client.
   */
  @Test (timeout = 120000)
  public void testSnapshotsForOpenFilesAndDeletion() throws Exception {
    // Construct the directory tree
    final Path snapRootDir = new Path("/level_0_A");
    final String flumeFileName = "flume.log";
    final String hbaseFileName = "hbase.log";
    final String snap1Name = "snap_1";
    final String snap2Name = "snap_2";
    final String snap3Name = "snap_3";

    // Create files and open streams
    final Path flumeFile = new Path(snapRootDir, flumeFileName);
    createFile(flumeFile);
    final Path hbaseFile = new Path(snapRootDir, hbaseFileName);
    createFile(hbaseFile);
    FSDataOutputStream flumeOutputStream = fs.append(flumeFile);
    FSDataOutputStream hbaseOutputStream = fs.append(hbaseFile);

    // Create Snapshot S1
    final Path snap1Dir = SnapshotTestHelper.createSnapshot(
        fs, snapRootDir, snap1Name);
    final Path flumeS1Path = new Path(snap1Dir, flumeFileName);
    final long flumeFileLengthAfterS1 = fs.getFileStatus(flumeFile).getLen();
    final Path hbaseS1Path = new Path(snap1Dir, hbaseFileName);
    final long hbaseFileLengthAfterS1 = fs.getFileStatus(hbaseFile).getLen();

    // Verify if Snap S1 file length is same as the the current versions
    Assert.assertEquals(flumeFileLengthAfterS1,
        fs.getFileStatus(flumeS1Path).getLen());
    Assert.assertEquals(hbaseFileLengthAfterS1,
        fs.getFileStatus(hbaseS1Path).getLen());

    long flumeFileWrittenDataLength = flumeFileLengthAfterS1;
    long hbaseFileWrittenDataLength = hbaseFileLengthAfterS1;
    int newWriteLength = (int) (BLOCKSIZE * 1.5);
    byte[] buf = new byte[newWriteLength];
    Random random = new Random();
    random.nextBytes(buf);

    // Write more data to files
    flumeFileWrittenDataLength += writeToStream(flumeOutputStream, buf);
    hbaseFileWrittenDataLength += writeToStream(hbaseOutputStream, buf);

    // Create Snapshot S2
    final Path snap2Dir = SnapshotTestHelper.createSnapshot(
        fs, snapRootDir, snap2Name);
    final Path flumeS2Path = new Path(snap2Dir, flumeFileName);
    final Path hbaseS2Path = new Path(snap2Dir, hbaseFileName);

    // Verify current files length are same as all data written till now
    final long flumeFileLengthAfterS2 = fs.getFileStatus(flumeFile).getLen();
    Assert.assertEquals(flumeFileWrittenDataLength, flumeFileLengthAfterS2);
    final long hbaseFileLengthAfterS2 = fs.getFileStatus(hbaseFile).getLen();
    Assert.assertEquals(hbaseFileWrittenDataLength, hbaseFileLengthAfterS2);

    // Verify if Snap S2 file length is same as the current versions
    Assert.assertEquals(flumeFileLengthAfterS2,
        fs.getFileStatus(flumeS2Path).getLen());
    Assert.assertEquals(hbaseFileLengthAfterS2,
        fs.getFileStatus(hbaseS2Path).getLen());

    // Write more data to open files
    writeToStream(flumeOutputStream, buf);
    hbaseFileWrittenDataLength += writeToStream(hbaseOutputStream, buf);

    // Verify old snapshots have point-in-time/frozen file
    // lengths even after the current versions have moved forward.
    Assert.assertEquals(flumeFileLengthAfterS1,
        fs.getFileStatus(flumeS1Path).getLen());
    Assert.assertEquals(flumeFileLengthAfterS2,
        fs.getFileStatus(flumeS2Path).getLen());
    Assert.assertEquals(hbaseFileLengthAfterS1,
        fs.getFileStatus(hbaseS1Path).getLen());
    Assert.assertEquals(hbaseFileLengthAfterS2,
        fs.getFileStatus(hbaseS2Path).getLen());

    // Delete flume current file. Snapshots should
    // still have references to flume file.
    boolean flumeFileDeleted = fs.delete(flumeFile, true);
    Assert.assertTrue(flumeFileDeleted);
    Assert.assertFalse(fs.exists(flumeFile));
    Assert.assertTrue(fs.exists(flumeS1Path));
    Assert.assertTrue(fs.exists(flumeS2Path));

    SnapshotTestHelper.createSnapshot(fs, snapRootDir, "tmp_snap");
    fs.deleteSnapshot(snapRootDir, "tmp_snap");

    // Delete snap_2. snap_1 still has reference to
    // the flume file.
    fs.deleteSnapshot(snapRootDir, snap2Name);
    Assert.assertFalse(fs.exists(flumeS2Path));
    Assert.assertTrue(fs.exists(flumeS1Path));

    // Delete snap_1. Now all traces of flume file
    // is gone.
    fs.deleteSnapshot(snapRootDir, snap1Name);
    Assert.assertFalse(fs.exists(flumeS2Path));
    Assert.assertFalse(fs.exists(flumeS1Path));

    // Create Snapshot S3
    final Path snap3Dir = SnapshotTestHelper.createSnapshot(
        fs, snapRootDir, snap3Name);
    final Path hbaseS3Path = new Path(snap3Dir, hbaseFileName);

    // Verify live files length is same as all data written till now
    final long hbaseFileLengthAfterS3 = fs.getFileStatus(hbaseFile).getLen();
    Assert.assertEquals(hbaseFileWrittenDataLength, hbaseFileLengthAfterS3);

    // Write more data to open files
    hbaseFileWrittenDataLength += writeToStream(hbaseOutputStream, buf);

    // Verify old snapshots have point-in-time/frozen file
    // lengths even after the flume open file is deleted and
    // the hbase live file has moved forward.
    Assert.assertEquals(hbaseFileLengthAfterS3,
        fs.getFileStatus(hbaseS3Path).getLen());
    Assert.assertEquals(hbaseFileWrittenDataLength,
        fs.getFileStatus(hbaseFile).getLen());

    hbaseOutputStream.close();
  }

  /**
   * Verify if the NameNode can restart properly after an OpenForWrite
   * file and the only snapshot it was present in were deleted.
   *
   * @throws Exception
   */
  @Test (timeout = 600000)
  public void testOpenFileDeletionAndNNRestart() throws Exception {
    // Construct the directory tree
    final Path snapRootDir = new Path("/level_0_A/test");
    final String hbaseFileName = "hbase.log";
    final String snap1Name = "snap_1";

    // Create a file with few blocks. Get its output stream
    // for append.
    final Path hbaseFile = new Path(snapRootDir, hbaseFileName);
    createFile(hbaseFile);
    FSDataOutputStream hbaseOutputStream = fs.append(hbaseFile);

    int newWriteLength = (int) (BLOCKSIZE * 1.5);
    byte[] buf = new byte[newWriteLength];
    Random random = new Random();
    random.nextBytes(buf);

    // Write more data to the file
    writeToStream(hbaseOutputStream, buf);

    // Take a snapshot while the file is open for write
    final Path snap1Dir = SnapshotTestHelper.createSnapshot(
        fs, snapRootDir, snap1Name);
    LOG.info("Open file status in snap: " +
        fs.getFileStatus(new Path(snap1Dir, hbaseFileName)));

    // Delete the open file and the snapshot while
    // its output stream is still open.
    fs.delete(hbaseFile, true);
    fs.deleteSnapshot(snapRootDir, snap1Name);
    Assert.assertFalse(fs.exists(hbaseFile));

    // Verify file existence after the NameNode restart
    cluster.restartNameNode();
    cluster.waitActive();
    Assert.assertFalse(fs.exists(hbaseFile));
  }

  /**
   * Test client writing to open files are not interrupted when snapshots
   * that captured open files get deleted.
   */
  @Test (timeout = 240000)
  public void testOpenFileWritingAcrossSnapDeletion() throws Exception {
    final Path snapRootDir = new Path("/level_0_A");
    final String flumeFileName = "flume.log";
    final String hbaseFileName = "hbase.log";
    final String snap1Name = "snap_1";
    final String snap2Name = "snap_2";
    final String snap3Name = "snap_3";

    // Create files and open streams
    final Path flumeFile = new Path(snapRootDir, flumeFileName);
    FSDataOutputStream flumeOut = fs.create(flumeFile, false,
        8000, (short)3, 1048576);
    flumeOut.close();
    final Path hbaseFile = new Path(snapRootDir, hbaseFileName);
    FSDataOutputStream hbaseOut = fs.create(hbaseFile, false,
        8000, (short)3, 1048576);
    hbaseOut.close();

    final AtomicBoolean writerError = new AtomicBoolean(false);
    final CountDownLatch startLatch = new CountDownLatch(1);
    final CountDownLatch deleteLatch = new CountDownLatch(1);
    Thread t = new Thread(new Runnable() {
      @Override
      public void run() {
        try {
          FSDataOutputStream flumeOutputStream = fs.append(flumeFile, 8000);
          FSDataOutputStream hbaseOutputStream = fs.append(hbaseFile, 8000);
          byte[] bytes = new byte[(int) (1024 * 0.2)];
          Random r = new Random(Time.now());

          for (int i = 0; i < 200000; i++) {
            r.nextBytes(bytes);
            flumeOutputStream.write(bytes);
            if (hbaseOutputStream != null) {
              hbaseOutputStream.write(bytes);
            }
            if (i == 50000) {
              startLatch.countDown();
            } else if (i == 100000) {
              deleteLatch.countDown();
            } else if (i == 150000) {
              hbaseOutputStream.hsync();
              fs.delete(hbaseFile, true);
              try {
                hbaseOutputStream.close();
              } catch (Exception e) {
                // since the file is deleted before the open stream close,
                // it might throw FileNotFoundException. Ignore the
                // expected exception.
              }
              hbaseOutputStream = null;
            } else if (i % 5000 == 0) {
              LOG.info("Write pos: " + flumeOutputStream.getPos()
                  + ", size: " + fs.getFileStatus(flumeFile).getLen()
                  + ", loop: " + (i + 1));
            }
          }
        } catch (Exception e) {
          LOG.warn("Writer error: " + e);
          writerError.set(true);
        }
      }
    });
    t.start();

    startLatch.await();
    final Path snap1Dir = SnapshotTestHelper.createSnapshot(
        fs, snapRootDir, snap1Name);
    final Path flumeS1Path = new Path(snap1Dir, flumeFileName);
    LOG.info("Snap1 file status: " + fs.getFileStatus(flumeS1Path));
    LOG.info("Current file status: " + fs.getFileStatus(flumeFile));

    deleteLatch.await();
    LOG.info("Snap1 file status: " + fs.getFileStatus(flumeS1Path));
    LOG.info("Current file status: " + fs.getFileStatus(flumeFile));

    // Verify deletion of snapshot which had the under construction file
    // captured is not truncating the under construction file and the thread
    // writing to the same file not crashing on newer block allocations.
    LOG.info("Deleting " + snap1Name);
    fs.deleteSnapshot(snapRootDir, snap1Name);

    // Verify creation and deletion of snapshot newer than the oldest
    // snapshot is not crashing the thread writing to under construction file.
    SnapshotTestHelper.createSnapshot(fs, snapRootDir, snap2Name);
    SnapshotTestHelper.createSnapshot(fs, snapRootDir, snap3Name);
    fs.deleteSnapshot(snapRootDir, snap3Name);
    fs.deleteSnapshot(snapRootDir, snap2Name);
    SnapshotTestHelper.createSnapshot(fs, snapRootDir, "test");

    t.join();
    Assert.assertFalse("Client encountered writing error!", writerError.get());

    restartNameNode();
    cluster.waitActive();
  }

  /**
   * Verify snapshots with open files captured are safe even when the
   * 'current' version of the file is truncated and appended later.
   */
  @Test (timeout = 120000)
  public void testOpenFilesSnapChecksumWithTrunkAndAppend() throws Exception {
    // Construct the directory tree
    final Path dir = new Path("/A/B/C");
    fs.mkdirs(dir);

    // String constants
    final Path hbaseSnapRootDir = dir;
    final String hbaseFileName = "hbase.wal";
    final String hbaseSnap1Name = "hbase_snap_s1";
    final String hbaseSnap2Name = "hbase_snap_s2";
    final String hbaseSnap3Name = "hbase_snap_s3";
    final String hbaseSnap4Name = "hbase_snap_s4";

    // Create files and open a stream
    final Path hbaseFile = new Path(dir, hbaseFileName);
    createFile(hbaseFile);
    final FileChecksum hbaseWALFileCksum0 =
        fs.getFileChecksum(hbaseFile);
    FSDataOutputStream hbaseOutputStream = fs.append(hbaseFile);

    // Create Snapshot S1
    final Path hbaseS1Dir = SnapshotTestHelper.createSnapshot(
        fs, hbaseSnapRootDir, hbaseSnap1Name);
    final Path hbaseS1Path = new Path(hbaseS1Dir, hbaseFileName);
    final FileChecksum hbaseFileCksumS1 = fs.getFileChecksum(hbaseS1Path);

    // Verify if Snap S1 checksum is same as the current version one
    Assert.assertEquals("Live and snap1 file checksum doesn't match!",
        hbaseWALFileCksum0, fs.getFileChecksum(hbaseS1Path));

    int newWriteLength = (int) (BLOCKSIZE * 1.5);
    byte[] buf = new byte[newWriteLength];
    Random random = new Random();
    random.nextBytes(buf);
    writeToStream(hbaseOutputStream, buf);

    // Create Snapshot S2
    final Path hbaseS2Dir = SnapshotTestHelper.createSnapshot(
        fs, hbaseSnapRootDir, hbaseSnap2Name);
    final Path hbaseS2Path = new Path(hbaseS2Dir, hbaseFileName);
    final FileChecksum hbaseFileCksumS2 = fs.getFileChecksum(hbaseS2Path);

    // Verify if the s1 checksum is still the same
    Assert.assertEquals("Snap file checksum has changed!",
        hbaseFileCksumS1, fs.getFileChecksum(hbaseS1Path));
    // Verify if the s2 checksum is different from the s1 checksum
    Assert.assertNotEquals("Snap1 and snap2 file checksum should differ!",
        hbaseFileCksumS1, hbaseFileCksumS2);

    newWriteLength = (int) (BLOCKSIZE * 2.5);
    buf = new byte[newWriteLength];
    random.nextBytes(buf);
    writeToStream(hbaseOutputStream, buf);

    // Create Snapshot S3
    final Path hbaseS3Dir = SnapshotTestHelper.createSnapshot(
        fs, hbaseSnapRootDir, hbaseSnap3Name);
    final Path hbaseS3Path = new Path(hbaseS3Dir, hbaseFileName);
    FileChecksum hbaseFileCksumS3 = fs.getFileChecksum(hbaseS3Path);

    // Record the checksum for the before truncate current file
    hbaseOutputStream.close();
    final FileChecksum hbaseFileCksumBeforeTruncate =
        fs.getFileChecksum(hbaseFile);
    Assert.assertEquals("Snap3 and before truncate file checksum should match!",
        hbaseFileCksumBeforeTruncate, hbaseFileCksumS3);

    // Truncate the current file and record the after truncate checksum
    long currentFileLen = fs.getFileStatus(hbaseFile).getLen();
    boolean fileTruncated = fs.truncate(hbaseFile, currentFileLen / 2);
    Assert.assertTrue("File truncation failed!", fileTruncated);
    final FileChecksum hbaseFileCksumAfterTruncate =
        fs.getFileChecksum(hbaseFile);

    Assert.assertNotEquals("Snap3 and after truncate checksum shouldn't match!",
        hbaseFileCksumS3, hbaseFileCksumAfterTruncate);

    // Append more data to the current file
    hbaseOutputStream = fs.append(hbaseFile);
    newWriteLength = (int) (BLOCKSIZE * 5.5);
    buf = new byte[newWriteLength];
    random.nextBytes(buf);
    writeToStream(hbaseOutputStream, buf);

    // Create Snapshot S4
    final Path hbaseS4Dir = SnapshotTestHelper.createSnapshot(
        fs, hbaseSnapRootDir, hbaseSnap4Name);
    final Path hbaseS4Path = new Path(hbaseS4Dir, hbaseFileName);
    final FileChecksum hbaseFileCksumS4 = fs.getFileChecksum(hbaseS4Path);

    // Record the checksum for the current file after append
    hbaseOutputStream.close();
    final FileChecksum hbaseFileCksumAfterAppend =
        fs.getFileChecksum(hbaseFile);

    Assert.assertEquals("Snap4 and after append file checksum should match!",
        hbaseFileCksumAfterAppend, hbaseFileCksumS4);

    // Recompute checksum for S3 path and verify it has not changed
    hbaseFileCksumS3 = fs.getFileChecksum(hbaseS3Path);
    Assert.assertEquals("Snap3 and before truncate file checksum should match!",
        hbaseFileCksumBeforeTruncate, hbaseFileCksumS3);
  }

  private Path createSnapshot(Path snapRootDir, String snapName,
      String fileName) throws Exception {
    final Path snap1Dir = SnapshotTestHelper.createSnapshot(
        fs, snapRootDir, snapName);
    return new Path(snap1Dir, fileName);
  }

  private void verifyFileSize(long fileSize, Path... filePaths) throws
      IOException {
    for (Path filePath : filePaths) {
      Assert.assertEquals(fileSize, fs.getFileStatus(filePath).getLen());
    }
  }

  /**
   * Verify open files captured in the snapshots across config disable
   * and enable.
   */
  @Test
  public void testOpenFilesWithMixedConfig() throws Exception {
    final Path snapRootDir = new Path("/level_0_A");
    final String flumeFileName = "flume.log";
    final String snap1Name = "s1";
    final String snap2Name = "s2";
    final String snap3Name = "s3";
    final String snap4Name = "s4";
    final String snap5Name = "s5";

    // Create files and open streams
    final Path flumeFile = new Path(snapRootDir, flumeFileName);
    createFile(flumeFile);
    FSDataOutputStream flumeOutputStream = fs.append(flumeFile);

    // 1. Disable capture open files
    cluster.getNameNode().getNamesystem()
        .getSnapshotManager().setCaptureOpenFiles(false);

    // Create Snapshot S1
    final Path flumeS1Path = createSnapshot(snapRootDir,
        snap1Name, flumeFileName);

    // Verify if Snap S1 file length is same as the the current versions
    verifyFileSize(FILELEN, flumeS1Path);

    // Write more data to files
    long flumeFileWrittenDataLength = FILELEN;
    int newWriteLength = (int) (BLOCKSIZE * 1.5);
    byte[] buf = new byte[newWriteLength];
    Random random = new Random();
    random.nextBytes(buf);
    flumeFileWrittenDataLength += writeToStream(flumeOutputStream, buf);

    // Create Snapshot S2
    final Path flumeS2Path = createSnapshot(snapRootDir,
        snap2Name, flumeFileName);

    // Since capture open files was disabled, all snapshots paths
    // and the current version should have same file lengths.
    verifyFileSize(flumeFileWrittenDataLength,
        flumeFile, flumeS2Path, flumeS1Path);

    // 2. Enable capture open files
    cluster.getNameNode().getNamesystem()
        .getSnapshotManager() .setCaptureOpenFiles(true);

    // Write more data to files
    flumeFileWrittenDataLength += writeToStream(flumeOutputStream, buf);
    long flumeFileLengthAfterS3 = flumeFileWrittenDataLength;

    // Create Snapshot S3
    final Path flumeS3Path = createSnapshot(snapRootDir,
        snap3Name, flumeFileName);

    // Since open files captured in the previous snapshots were with config
    // disabled, their file lengths are now same as the current version.
    // With the config turned on, any new data written to the open files
    // will no more reflect in the current version or old snapshot paths.
    verifyFileSize(flumeFileWrittenDataLength, flumeFile, flumeS3Path,
        flumeS2Path, flumeS1Path);

    // Write more data to files
    flumeFileWrittenDataLength += writeToStream(flumeOutputStream, buf);

    // Create Snapshot S4
    final Path flumeS4Path = createSnapshot(snapRootDir,
        snap4Name, flumeFileName);

    // Verify S4 has the latest data
    verifyFileSize(flumeFileWrittenDataLength, flumeFile, flumeS4Path);

    // But, open files captured as of Snapshot S3 and before should
    // have their old file lengths intact.
    verifyFileSize(flumeFileLengthAfterS3, flumeS3Path,
        flumeS2Path, flumeS1Path);

    long flumeFileLengthAfterS4 =  flumeFileWrittenDataLength;

    // 3. Disable capture open files
    cluster.getNameNode().getNamesystem()
        .getSnapshotManager() .setCaptureOpenFiles(false);

    // Create Snapshot S5
    final Path flumeS5Path = createSnapshot(snapRootDir,
        snap5Name, flumeFileName);

    flumeFileWrittenDataLength += writeToStream(flumeOutputStream, buf);

    // Since capture open files was disabled, any snapshots taken after the
    // config change and the current version should have same file lengths
    // for the open files.
    verifyFileSize(flumeFileWrittenDataLength, flumeFile, flumeS5Path);

    // But, the old snapshots taken before the config disable should
    // continue to be consistent.
    verifyFileSize(flumeFileLengthAfterS4, flumeS4Path);
  }

  private void restartNameNode() throws Exception {
    cluster.triggerBlockReports();
    NameNode nameNode = cluster.getNameNode();
    NameNodeAdapter.enterSafeMode(nameNode, false);
    NameNodeAdapter.saveNamespace(nameNode);
    NameNodeAdapter.leaveSafeMode(nameNode);
    cluster.restartNameNode(true);
  }
}