/**
 * 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.datanode.fsdataset.impl;

import com.google.common.base.Supplier;
import org.apache.commons.io.IOUtils;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
import static org.apache.hadoop.hdfs.DFSConfigKeys.*;
import static org.hamcrest.core.Is.is;
import static org.junit.Assert.assertThat;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.fail;

import org.apache.hadoop.fs.BlockLocation;
import org.apache.hadoop.fs.CommonConfigurationKeysPublic;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.HdfsBlockLocation;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hdfs.*;
import org.apache.hadoop.hdfs.protocol.Block;
import org.apache.hadoop.hdfs.protocol.LocatedBlock;
import org.apache.hadoop.hdfs.server.datanode.DataNode;
import org.apache.hadoop.hdfs.server.datanode.DataNodeFaultInjector;
import org.apache.hadoop.hdfs.server.datanode.fsdataset.FsDatasetSpi;
import org.apache.hadoop.hdfs.server.datanode.fsdataset.FsVolumeReference;
import org.apache.hadoop.ipc.RemoteException;
import org.apache.hadoop.test.GenericTestUtils;
import org.apache.hadoop.util.Daemon;
import org.apache.log4j.Level;
import org.junit.After;
import org.junit.Before;
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.ExpectedException;
import org.mockito.Mockito;

import java.io.IOException;
import java.io.OutputStream;
import java.lang.management.ManagementFactory;
import java.lang.reflect.Field;
import java.util.Map;
import java.util.Random;
import java.util.concurrent.TimeoutException;

import javax.management.MBeanServer;
import javax.management.ObjectName;

/**
 * Ensure that the DN reserves disk space equivalent to a full block for
 * replica being written (RBW) & Replica being copied from another DN.
 */
public class TestSpaceReservation {
  static final Log LOG = LogFactory.getLog(TestSpaceReservation.class);

  private static final int DU_REFRESH_INTERVAL_MSEC = 500;
  private static final int STORAGES_PER_DATANODE = 1;
  private static final int BLOCK_SIZE = 1024 * 1024;
  private static final int SMALL_BLOCK_SIZE = 1024;

  protected MiniDFSCluster cluster;
  private Configuration conf;
  private DistributedFileSystem fs = null;
  private DFSClient client = null;
  FsVolumeReference singletonVolumeRef = null;
  FsVolumeImpl singletonVolume = null;
  private DataNodeFaultInjector old = null;

  private static Random rand = new Random();

  @Before
  public void before() {
    conf = new HdfsConfiguration();
  }

  private void initConfig(int blockSize) {
    // Refresh disk usage information frequently.
    conf.setInt(FS_DU_INTERVAL_KEY, DU_REFRESH_INTERVAL_MSEC);
    conf.setLong(DFS_BLOCK_SIZE_KEY, blockSize);

    // Disable the scanner
    conf.setInt(DFS_DATANODE_SCAN_PERIOD_HOURS_KEY, -1);
  }

  static {
    GenericTestUtils.setLogLevel(FsDatasetImpl.LOG, Level.ALL);
    GenericTestUtils.setLogLevel(DataNode.LOG, Level.ALL);
  }

  /**
   *
   * @param blockSize
   * @param perVolumeCapacity limit the capacity of each volume to the given
   *                          value. If negative, then don't limit.
   * @throws IOException
   */
  private void startCluster(int blockSize, int numDatanodes, long perVolumeCapacity) throws IOException {
    initConfig(blockSize);

    cluster = new MiniDFSCluster
        .Builder(conf)
        .storagesPerDatanode(STORAGES_PER_DATANODE)
        .numDataNodes(numDatanodes)
        .build();
    fs = cluster.getFileSystem();
    client = fs.getClient();
    cluster.waitActive();

    if (perVolumeCapacity >= 0) {
      try (FsDatasetSpi.FsVolumeReferences volumes =
          cluster.getDataNodes().get(0).getFSDataset().getFsVolumeReferences()) {
        singletonVolumeRef = volumes.get(0).obtainReference();
      }
      singletonVolume = ((FsVolumeImpl) singletonVolumeRef.getVolume());
      singletonVolume.setCapacityForTesting(perVolumeCapacity);
    }
  }

  @After
  public void shutdownCluster() throws IOException {
    if (singletonVolumeRef != null) {
      singletonVolumeRef.close();
      singletonVolumeRef = null;
    }

    if (client != null) {
      client.close();
      client = null;
    }

    if (fs != null) {
      fs.close();
      fs = null;
    }

    if (cluster != null) {
      cluster.shutdown();
      cluster = null;
    }
    if (old != null) {
      DataNodeFaultInjector.set(old);
    }
  }

  private void createFileAndTestSpaceReservation(
      final String fileNamePrefix, final int fileBlockSize)
      throws IOException, InterruptedException {
    // Enough for 1 block + meta files + some delta.
    final long configuredCapacity = fileBlockSize * 2 - 1;
    startCluster(BLOCK_SIZE, 1, configuredCapacity);
    FSDataOutputStream out = null;
    Path path = new Path("/" + fileNamePrefix + ".dat");

    try {
      out = fs.create(path, false, 4096, (short) 1, fileBlockSize);

      byte[] buffer = new byte[rand.nextInt(fileBlockSize / 4)];
      out.write(buffer);
      out.hsync();
      int bytesWritten = buffer.length;

      // Check that space was reserved for a full block minus the bytesWritten.
      assertThat(singletonVolume.getReservedForReplicas(),
                 is((long) fileBlockSize - bytesWritten));
      out.close();
      out = null;

      // Check that the reserved space has been released since we closed the
      // file.
      assertThat(singletonVolume.getReservedForReplicas(), is(0L));

      // Reopen the file for appends and write 1 more byte.
      out = fs.append(path);
      out.write(buffer);
      out.hsync();
      bytesWritten += buffer.length;

      // Check that space was again reserved for a full block minus the
      // bytesWritten so far.
      assertThat(singletonVolume.getReservedForReplicas(),
                 is((long) fileBlockSize - bytesWritten));

      // Write once again and again verify the available space. This ensures
      // that the reserved space is progressively adjusted to account for bytes
      // written to disk.
      out.write(buffer);
      out.hsync();
      bytesWritten += buffer.length;
      assertThat(singletonVolume.getReservedForReplicas(),
                 is((long) fileBlockSize - bytesWritten));
    } finally {
      if (out != null) {
        out.close();
      }
    }
  }

  @Test (timeout=300000)
  public void testWithDefaultBlockSize()
      throws IOException, InterruptedException {
    createFileAndTestSpaceReservation(GenericTestUtils.getMethodName(), BLOCK_SIZE);
  }

  @Test (timeout=300000)
  public void testWithNonDefaultBlockSize()
      throws IOException, InterruptedException {
    // Same test as previous one, but with a non-default block size.
    createFileAndTestSpaceReservation(GenericTestUtils.getMethodName(), BLOCK_SIZE * 2);
  }

  @Rule
  public ExpectedException thrown = ExpectedException.none();

  @Test (timeout=300000)
  public void testWithLimitedSpace() throws IOException {
    // Cluster with just enough space for a full block + meta.
    startCluster(BLOCK_SIZE, 1, 2 * BLOCK_SIZE - 1);
    final String methodName = GenericTestUtils.getMethodName();
    Path file1 = new Path("/" + methodName + ".01.dat");
    Path file2 = new Path("/" + methodName + ".02.dat");

    // Create two files.
    FSDataOutputStream os1 = null, os2 = null;

    try {
      os1 = fs.create(file1);
      os2 = fs.create(file2);

      // Write one byte to the first file.
      byte[] data = new byte[1];
      os1.write(data);
      os1.hsync();

      // Try to write one byte to the second file.
      // The block allocation must fail.
      thrown.expect(RemoteException.class);
      os2.write(data);
      os2.hsync();
    } finally {
      if (os1 != null) {
        os1.close();
      }

      // os2.close() will fail as no block was allocated.
    }
  }

  /**
   * Ensure that reserved space is released when the client goes away
   * unexpectedly.
   *
   * The verification is done for each replica in the write pipeline.
   *
   * @throws IOException
   */
  @Test(timeout=300000)
  public void testSpaceReleasedOnUnexpectedEof()
      throws IOException, InterruptedException, TimeoutException {
    final short replication = 3;
    startCluster(BLOCK_SIZE, replication, -1);

    final String methodName = GenericTestUtils.getMethodName();
    final Path file = new Path("/" + methodName + ".01.dat");

    // Write 1 byte to the file and kill the writer.
    FSDataOutputStream os = fs.create(file, replication);
    os.write(new byte[1]);
    os.hsync();
    DFSTestUtil.abortStream((DFSOutputStream) os.getWrappedStream());

    // Ensure all space reserved for the replica was released on each
    // DataNode.
    for (DataNode dn : cluster.getDataNodes()) {
      try (FsDatasetSpi.FsVolumeReferences volumes =
          dn.getFSDataset().getFsVolumeReferences()) {
        final FsVolumeImpl volume = (FsVolumeImpl) volumes.get(0);
        GenericTestUtils.waitFor(new Supplier<Boolean>() {
          @Override
          public Boolean get() {
            return (volume.getReservedForReplicas() == 0);
          }
        }, 500, Integer.MAX_VALUE); // Wait until the test times out.
      }
    }
  }

  @SuppressWarnings("unchecked")
  @Test(timeout = 30000)
  public void testRBWFileCreationError() throws Exception {

    final short replication = 1;
    startCluster(BLOCK_SIZE, replication, -1);

    final FsVolumeImpl fsVolumeImpl = (FsVolumeImpl) cluster.getDataNodes()
        .get(0).getFSDataset().getFsVolumeReferences().get(0);
    final String methodName = GenericTestUtils.getMethodName();
    final Path file = new Path("/" + methodName + ".01.dat");

    // Mock BlockPoolSlice so that RBW file creation gives IOExcception
    BlockPoolSlice blockPoolSlice = Mockito.mock(BlockPoolSlice.class);
    Mockito.when(blockPoolSlice.createRbwFile((Block) Mockito.any()))
        .thenThrow(new IOException("Synthetic IO Exception Throgh MOCK"));

    Field field = FsVolumeImpl.class.getDeclaredField("bpSlices");
    field.setAccessible(true);
    Map<String, BlockPoolSlice> bpSlices = (Map<String, BlockPoolSlice>) field
        .get(fsVolumeImpl);
    bpSlices.put(fsVolumeImpl.getBlockPoolList()[0], blockPoolSlice);

    try {
      // Write 1 byte to the file
      FSDataOutputStream os = fs.create(file, replication);
      os.write(new byte[1]);
      os.hsync();
      os.close();
      fail("Expecting IOException file creation failure");
    } catch (IOException e) {
      // Exception can be ignored (expected)
    }

    // Ensure RBW space reserved is released
    assertTrue(
        "Expected ZERO but got " + fsVolumeImpl.getReservedForReplicas(),
        fsVolumeImpl.getReservedForReplicas() == 0);

    // Reserve some bytes to verify double clearing space should't happen
    fsVolumeImpl.reserveSpaceForReplica(1000);
    try {
      // Write 1 byte to the file
      FSDataOutputStream os = fs.create(new Path("/" + methodName + ".02.dat"),
          replication);
      os.write(new byte[1]);
      os.hsync();
      os.close();
      fail("Expecting IOException file creation failure");
    } catch (IOException e) {
      // Exception can be ignored (expected)
    }

    // Ensure RBW space reserved is released only once
    assertTrue(fsVolumeImpl.getReservedForReplicas() == 1000);
  }

  @Test(timeout = 30000)
  public void testReservedSpaceInJMXBean() throws Exception {

    final short replication = 1;
    startCluster(BLOCK_SIZE, replication, -1);

    final String methodName = GenericTestUtils.getMethodName();
    final Path file = new Path("/" + methodName + ".01.dat");

    try (FSDataOutputStream os = fs.create(file, replication)) {
      // Write 1 byte to the file
      os.write(new byte[1]);
      os.hsync();

      final MBeanServer mbs = ManagementFactory.getPlatformMBeanServer();
      final ObjectName mxbeanName = new ObjectName(
          "Hadoop:service=DataNode,name=DataNodeInfo");
      final String volumeInfo = (String) mbs.getAttribute(mxbeanName,
          "VolumeInfo");

      // verify reserved space for Replicas in JMX bean volume info
      assertTrue(volumeInfo.contains("reservedSpaceForReplicas"));
    }
  }

  @Test(timeout = 300000)
  public void testTmpSpaceReserve() throws Exception {

    final short replication = 2;
    startCluster(BLOCK_SIZE, replication, -1);
    final int byteCount1 = 100;
    final int byteCount2 = 200;

    final String methodName = GenericTestUtils.getMethodName();

    // Test positive scenario
    {
      final Path file = new Path("/" + methodName + ".01.dat");

      try (FSDataOutputStream os = fs.create(file, (short) 1)) {
        // Write test data to the file
        os.write(new byte[byteCount1]);
        os.hsync();
      }

      BlockLocation[] blockLocations = fs.getFileBlockLocations(file, 0, 10);
      String firstReplicaNode = blockLocations[0].getNames()[0];

      int newReplicaDNIndex = 0;
      if (firstReplicaNode.equals(cluster.getDataNodes().get(0)
          .getDisplayName())) {
        newReplicaDNIndex = 1;
      }

      FsVolumeImpl fsVolumeImpl = (FsVolumeImpl) cluster.getDataNodes()
          .get(newReplicaDNIndex).getFSDataset().getFsVolumeReferences().get(0);

      performReReplication(file, true);

      assertEquals("Wrong reserve space for Tmp ", byteCount1,
          fsVolumeImpl.getRecentReserved());

      assertEquals("Reserved Tmp space is not released", 0,
          fsVolumeImpl.getReservedForReplicas());
    }

    // Test when file creation fails
    {
      final Path file = new Path("/" + methodName + ".01.dat");

      try (FSDataOutputStream os = fs.create(file, (short) 1)) {
        // Write test data to the file
        os.write(new byte[byteCount2]);
        os.hsync();
      }

      BlockLocation[] blockLocations = fs.getFileBlockLocations(file, 0, 10);
      String firstReplicaNode = blockLocations[0].getNames()[0];

      int newReplicaDNIndex = 0;
      if (firstReplicaNode.equals(cluster.getDataNodes().get(0)
          .getDisplayName())) {
        newReplicaDNIndex = 1;
      }

      BlockPoolSlice blockPoolSlice = Mockito.mock(BlockPoolSlice.class);
      Mockito.when(blockPoolSlice.createTmpFile((Block) Mockito.any()))
          .thenThrow(new IOException("Synthetic IO Exception Throgh MOCK"));

      final FsVolumeImpl fsVolumeImpl = (FsVolumeImpl) cluster.getDataNodes()
          .get(newReplicaDNIndex).getFSDataset().getFsVolumeReferences().get(0);

      // Reserve some bytes to verify double clearing space should't happen
      fsVolumeImpl.reserveSpaceForReplica(1000);

      Field field = FsVolumeImpl.class.getDeclaredField("bpSlices");
      field.setAccessible(true);
      @SuppressWarnings("unchecked")
      Map<String, BlockPoolSlice> bpSlices = (Map<String, BlockPoolSlice>) field
          .get(fsVolumeImpl);
      bpSlices.put(fsVolumeImpl.getBlockPoolList()[0], blockPoolSlice);

      performReReplication(file, false);

      assertEquals("Wrong reserve space for Tmp ", byteCount2,
          fsVolumeImpl.getRecentReserved());

      assertEquals("Tmp space is not released OR released twice", 1000,
          fsVolumeImpl.getReservedForReplicas());
    }
  }

  private void performReReplication(Path filePath, boolean waitForSuccess)
      throws Exception {
    fs.setReplication(filePath, (short) 2);

    Thread.sleep(4000);
    BlockLocation[] blockLocations = fs.getFileBlockLocations(filePath, 0, 10);

    if (waitForSuccess) {
      // Wait for the re replication
      while (blockLocations[0].getNames().length < 2) {
        Thread.sleep(2000);
        blockLocations = fs.getFileBlockLocations(filePath, 0, 10);
      }
    }
  }

  /**
   * Stress test to ensure we are not leaking reserved space.
   * @throws IOException
   * @throws InterruptedException
   */
  @Test (timeout=600000)
  public void stressTest() throws IOException, InterruptedException {
    final int numWriters = 5;
    startCluster(SMALL_BLOCK_SIZE, 1, SMALL_BLOCK_SIZE * numWriters * 10);
    Writer[] writers = new Writer[numWriters];

    // Start a few writers and let them run for a while.
    for (int i = 0; i < numWriters; ++i) {
      writers[i] = new Writer(client, SMALL_BLOCK_SIZE);
      writers[i].start();
    }

    Thread.sleep(60000);

    // Stop the writers.
    for (Writer w : writers) {
      w.stopWriter();
    }
    int filesCreated = 0;
    int numFailures = 0;
    for (Writer w : writers) {
      w.join();
      filesCreated += w.getFilesCreated();
      numFailures += w.getNumFailures();
    }

    LOG.info("Stress test created " + filesCreated +
             " files and hit " + numFailures + " failures");

    // Check no space was leaked.
    assertThat(singletonVolume.getReservedForReplicas(), is(0L));
  }

  private static class Writer extends Daemon {
    private volatile boolean keepRunning;
    private final DFSClient localClient;
    private int filesCreated = 0;
    private int numFailures = 0;
    byte[] data;

    Writer(DFSClient client, int blockSize) throws IOException {
      localClient = client;
      keepRunning = true;
      filesCreated = 0;
      numFailures = 0;

      // At least some of the files should span a block boundary.
      data = new byte[blockSize * 2];
    }

    @Override
    public void run() {
      /**
       * Create a file, write up to 3 blocks of data and close the file.
       * Do this in a loop until we are told to stop.
       */
      while (keepRunning) {
        OutputStream os = null;
        try {
          String filename = "/file-" + rand.nextLong();
          os = localClient.create(filename, false);
          os.write(data, 0, rand.nextInt(data.length));
          IOUtils.closeQuietly(os);
          os = null;
          localClient.delete(filename, false);
          Thread.sleep(50);     // Sleep for a bit to avoid killing the system.
          ++filesCreated;
        } catch (IOException ioe) {
          // Just ignore the exception and keep going.
          ++numFailures;
        } catch (InterruptedException ie) {
          return;
        } finally {
          if (os != null) {
            IOUtils.closeQuietly(os);
          }
        }
      }
    }

    public void stopWriter() {
      keepRunning = false;
    }

    public int getFilesCreated() {
      return filesCreated;
    }

    public int getNumFailures() {
      return numFailures;
    }
  }

  @Test(timeout = 30000)
  public void testReservedSpaceForAppend() throws Exception {
    final short replication = 3;
    startCluster(BLOCK_SIZE, replication, -1);
    final String methodName = GenericTestUtils.getMethodName();
    final Path file = new Path("/" + methodName + ".01.dat");

    // Write 1 byte to the file and kill the writer.
    FSDataOutputStream os = fs.create(file, replication);
    os.write(new byte[1024]);
    os.close();

    final Path file2 = new Path("/" + methodName + ".02.dat");

    // Write 1 byte to the file and keep it open.
    FSDataOutputStream os2 = fs.create(file2, replication);
    os2.write(new byte[1]);
    os2.hflush();
    int expectedFile2Reserved = BLOCK_SIZE - 1;
    checkReservedSpace(expectedFile2Reserved);

    // append one byte and verify reservedspace before and after closing
    os = fs.append(file);
    os.write(new byte[1]);
    os.hflush();
    int expectedFile1Reserved = BLOCK_SIZE - 1025;
    checkReservedSpace(expectedFile2Reserved + expectedFile1Reserved);
    os.close();
    checkReservedSpace(expectedFile2Reserved);

    // append one byte and verify reservedspace before and after abort
    os = fs.append(file);
    os.write(new byte[1]);
    os.hflush();
    expectedFile1Reserved--;
    checkReservedSpace(expectedFile2Reserved + expectedFile1Reserved);
    DFSTestUtil.abortStream(((DFSOutputStream) os.getWrappedStream()));
    checkReservedSpace(expectedFile2Reserved);
  }

  @Test(timeout = 30000)
  public void testReservedSpaceForPipelineRecovery() throws Exception {
    final short replication = 3;
    startCluster(BLOCK_SIZE, replication, -1);

    final String methodName = GenericTestUtils.getMethodName();
    final Path file = new Path("/" + methodName + ".01.dat");

    old = DataNodeFaultInjector.get();
    // Fault injector to fail connection to mirror first time.
    DataNodeFaultInjector.set(new DataNodeFaultInjector() {
      private int tries = 0;

      @Override
      public void failMirrorConnection() throws IOException {
        if (tries++ == 0) {
          throw new IOException("Failing Mirror for space reservation");
        }
      }
    });
    // Write 1 byte to the file and kill the writer.
    FSDataOutputStream os = fs.create(file, replication);
    os.write(new byte[1]);
    os.close();
    // Ensure all space reserved for the replica was released on each
    // DataNode.
    cluster.triggerBlockReports();
    for (final DataNode dn : cluster.getDataNodes()) {
      try (FsDatasetSpi.FsVolumeReferences volumes =
          dn.getFSDataset().getFsVolumeReferences()) {
        final FsVolumeImpl volume = (FsVolumeImpl) volumes.get(0);
        GenericTestUtils.waitFor(new Supplier<Boolean>() {
          @Override
          public Boolean get() {
            LOG.info("dn " + dn.getDisplayName() + " space : "
                + volume.getReservedForReplicas());
            return (volume.getReservedForReplicas() == 0);
          }
        }, 100, Integer.MAX_VALUE); // Wait until the test times out.
      }
    }
  }

  private void checkReservedSpace(final long expectedReserved) throws TimeoutException,
      InterruptedException, IOException {
    for (final DataNode dn : cluster.getDataNodes()) {
      try (FsDatasetSpi.FsVolumeReferences volumes = dn.getFSDataset()
          .getFsVolumeReferences()) {
        final FsVolumeImpl volume = (FsVolumeImpl) volumes.get(0);
        GenericTestUtils.waitFor(new Supplier<Boolean>() {
          @Override
          public Boolean get() {
            LOG.info(
                "dn " + dn.getDisplayName() + " space : " + volume
                    .getReservedForReplicas() + ", Expected ReservedSpace :"
                    + expectedReserved);
            return (volume.getReservedForReplicas() == expectedReserved);
          }
        }, 100, 3000);
      }
    }
  }

  @Test(timeout = 60000)
  public void testReservedSpaceForLeaseRecovery() throws Exception {
    final short replication = 3;
    conf.setInt(
        CommonConfigurationKeysPublic.IPC_CLIENT_CONNECT_MAX_RETRIES_KEY, 2);
    conf.setInt(
        CommonConfigurationKeysPublic.IPC_CLIENT_CONNECT_RETRY_INTERVAL_KEY,
        1000);
    startCluster(BLOCK_SIZE, replication, -1);

    final String methodName = GenericTestUtils.getMethodName();
    final Path file = new Path("/" + methodName + ".01.dat");
    // Write to the file and kill the writer.
    FSDataOutputStream os = fs.create(file, replication);
    os.write(new byte[8192]);
    os.hflush();
    os.close();
    /*
     * Reset the pipeline for the append in such a way that, datanode which is
     * down is one of the mirror, not the first datanode.
     */
    HdfsBlockLocation blockLocation = (HdfsBlockLocation) fs.getClient()
        .getBlockLocations(file.toString(), 0, BLOCK_SIZE)[0];
    LocatedBlock lastBlock = blockLocation.getLocatedBlock();
    // stop 3rd node.
    cluster.stopDataNode(lastBlock.getLocations()[2].getName());
    try {
      os = fs.append(file);
      DFSTestUtil.setPipeline((DFSOutputStream) os.getWrappedStream(),
          lastBlock);
      os.writeBytes("hi");
      os.hsync();
    } catch (IOException e) {
      // Append will fail due to not able to replace datanodes in 3 nodes
      // cluster.
      LOG.info("", e);
    }
    DFSTestUtil.abortStream((DFSOutputStream) os.getWrappedStream());
    /*
     * There is a chance that stopped DN could be chosen as primary for
     * recovery. If so, then recovery will not happen in time. So mark stopped
     * node as dead to exclude that node.
     */
    cluster.setDataNodeDead(lastBlock.getLocations()[2]);
    fs.recoverLease(file);
    GenericTestUtils.waitFor(new Supplier<Boolean>() {
      @Override
      public Boolean get() {
        try {
          return fs.isFileClosed(file);
        } catch (IOException e) {
          return false;
        }
      }
    }, 500, 30000);
    checkReservedSpace(0);
  }
}
