@Override  // FsDatasetSpi
  public ReplicaHandler append(ExtendedBlock b,
      long newGS, long expectedBlockLen) throws IOException {
    try (AutoCloseableLock lock = lockManager.writeLock(LockLevel.VOLUME,
        b.getBlockPoolId(), getStorageUuidForLock(b))) {
      // If the block was successfully finalized because all packets
      // were successfully processed at the Datanode but the ack for
      // some of the packets were not received by the client. The client
      // re-opens the connection and retries sending those packets.
      // The other reason is that an "append" is occurring to this block.

      // check the validity of the parameter
      if (newGS < b.getGenerationStamp()) {
        throw new IOException("The new generation stamp " + newGS +
            " should be greater than the replica " + b + "'s generation stamp");
      }
      ReplicaInfo replicaInfo = getReplicaInfo(b);
      LOG.info("Appending to " + replicaInfo);
      if (replicaInfo.getState() != ReplicaState.FINALIZED) {
        throw new ReplicaNotFoundException(
            ReplicaNotFoundException.UNFINALIZED_REPLICA + b);
      }
      if (replicaInfo.getNumBytes() != expectedBlockLen) {
        throw new IOException("Corrupted replica " + replicaInfo +
            " with a length of " + replicaInfo.getNumBytes() +
            " expected length is " + expectedBlockLen);
      }

      FsVolumeReference ref = replicaInfo.getVolume().obtainReference();
      ReplicaInPipeline replica = null;
      try {
        replica = append(b.getBlockPoolId(), replicaInfo, newGS,
            b.getNumBytes());
      } catch (IOException e) {
        IOUtils.cleanupWithLogger(null, ref);
        throw e;
      }
      return new ReplicaHandler(replica, ref);
    }
  }
@Test(timeout = 30000)
  public void testAppend() {
    MiniDFSCluster cluster = null;
    try {
      cluster = new MiniDFSCluster.Builder(conf)
          .numDataNodes(1)
          .storageTypes(new StorageType[]{StorageType.DISK, StorageType.DISK})
          .storagesPerDatanode(2)
          .build();
      FileSystem fs = cluster.getFileSystem();
      DataNode dataNode = cluster.getDataNodes().get(0);

      // Create test file
      Path filePath = new Path("testData");
      FsDatasetImpl fsDataSetImpl = (FsDatasetImpl) dataNode.getFSDataset();
      DFSTestUtil.createFile(fs, filePath, 100, (short) 1, 0);
      ExtendedBlock block = DFSTestUtil.getFirstBlock(fs, filePath);
      ReplicaInfo replicaInfo = fsDataSetImpl.getReplicaInfo(block);
      long oldMetaLength = replicaInfo.getMetadataLength();
      long oldDfsUsed = fsDataSetImpl.getDfsUsed();

      // Append to file
      int appendLength = 100;
      DFSTestUtil.appendFile(fs, filePath, appendLength);

      block = DFSTestUtil.getFirstBlock(fs, filePath);
      replicaInfo = fsDataSetImpl.getReplicaInfo(block);
      long newMetaLength = replicaInfo.getMetadataLength();
      long newDfsUsed = fsDataSetImpl.getDfsUsed();

      assert newDfsUsed == oldDfsUsed + appendLength + (newMetaLength - oldMetaLength) :
          "When appending a file, the dfsused statistics of datanode are incorrect.";
    } catch (Exception ex) {
      LOG.info("Exception in testAppend ", ex);
      fail("Exception while testing testAppend ");
    } finally {
      if (cluster.isClusterUp()) {
        cluster.shutdown();
      }
    }
  }