import org.apache.hadoop.hdfs.server.common.HdfsServerConstants.BlockUCState; //导入依赖的package包/类
@Override // BlockCollection, the file should be under construction
public BlockInfoContiguousUnderConstruction setLastBlock(
    BlockInfoContiguous lastBlock, DatanodeStorageInfo[] locations)
    throws IOException {
  Preconditions.checkState(isUnderConstruction(),
      "file is no longer under construction");

  if (numBlocks() == 0) {
    throw new IOException("Failed to set last block: File is empty.");
  }
  BlockInfoContiguousUnderConstruction ucBlock =
    lastBlock.convertToBlockUnderConstruction(
        BlockUCState.UNDER_CONSTRUCTION, locations);
  setBlock(numBlocks() - 1, ucBlock);
  return ucBlock;
}
 

import org.apache.hadoop.hdfs.server.common.HdfsServerConstants.BlockUCState; //导入依赖的package包/类
private boolean isBlockUnderConstruction(BlockInfoContiguous storedBlock,
    BlockUCState ucState, ReplicaState reportedState) {
  switch(reportedState) {
  case FINALIZED:
    switch(ucState) {
    case UNDER_CONSTRUCTION:
    case UNDER_RECOVERY:
      return true;
    default:
      return false;
    }
  case RBW:
  case RWR:
    return (!storedBlock.isComplete());
  case RUR:       // should not be reported                                                                                             
  case TEMPORARY: // should not be reported                                                                                             
  default:
    return false;
  }
}
 

import org.apache.hadoop.hdfs.server.common.HdfsServerConstants.BlockUCState; //导入依赖的package包/类
private boolean isBlockUnderConstruction(BlockInfo storedBlock,
    BlockUCState ucState, ReplicaState reportedState) {
  switch(reportedState) {
  case FINALIZED:
    switch(ucState) {
    case UNDER_CONSTRUCTION:
    case UNDER_RECOVERY:
      return true;
    default:
      return false;
    }
  case RBW:
  case RWR:
    return (!storedBlock.isComplete());
  case RUR:       // should not be reported                                                                                             
  case TEMPORARY: // should not be reported                                                                                             
  default:
    return false;
  }
}
 

import org.apache.hadoop.hdfs.server.common.HdfsServerConstants.BlockUCState; //导入依赖的package包/类
/**
 * Check that the indicated blocks are present and
 * replicated.
 */
public boolean checkBlocksProperlyReplicated(
    String src, BlockInfo[] blocks) {
  for (BlockInfo b: blocks) {
    if (!b.isComplete()) {
      final int numNodes = b.numNodes();
      final int min = getMinStorageNum(b);
      final BlockUCState state = b.getBlockUCState();
      LOG.info("BLOCK* " + b + " is not COMPLETE (ucState = " + state
          + ", replication# = " + numNodes + (numNodes < min ? " < " : " >= ")
          + " minimum = " + min + ") in file " + src);
      return false;
    }
  }
  return true;
}
 

import org.apache.hadoop.hdfs.server.common.HdfsServerConstants.BlockUCState; //导入依赖的package包/类
@Override // BlockCollection, the file should be under construction
public BlockInfoUnderConstruction setLastBlock(BlockInfo lastBlock,
    DatanodeStorageInfo[] locations) throws IOException {
  Preconditions.checkState(isUnderConstruction(),
      "file is no longer under construction");

  if (numBlocks() == 0) {
    throw new IOException("Failed to set last block: File is empty.");
  }
  BlockInfoUnderConstruction ucBlock =
    lastBlock.convertToBlockUnderConstruction(
        BlockUCState.UNDER_CONSTRUCTION, locations);
  ucBlock.setBlockCollection(this);
  setBlock(numBlocks() - 1, ucBlock);
  return ucBlock;
}
 

import org.apache.hadoop.hdfs.server.common.HdfsServerConstants.BlockUCState; //导入依赖的package包/类
private boolean isBlockUnderConstruction(BlockInfo storedBlock, 
    BlockUCState ucState, ReplicaState reportedState) {
  switch(reportedState) {
  case FINALIZED:
    switch(ucState) {
    case UNDER_CONSTRUCTION:
    case UNDER_RECOVERY:
      return true;
    default:
      return false;
    }
  case RBW:
  case RWR:
    return (!storedBlock.isComplete());
  case RUR:       // should not be reported                                                                                             
  case TEMPORARY: // should not be reported                                                                                             
  default:
    return false;
  }
}
 

import org.apache.hadoop.hdfs.server.common.HdfsServerConstants.BlockUCState; //导入依赖的package包/类
private boolean isBlockUnderConstruction(BlockInfo storedBlock,
    BlockUCState ucState, ReplicaState reportedState) {
  switch (reportedState) {
    case FINALIZED:
      switch (ucState) {
        case UNDER_CONSTRUCTION:
        case UNDER_RECOVERY:
          return true;
        default:
          return false;
      }
    case RBW:
    case RWR:
      return (!storedBlock.isComplete());
    case RUR:       // should not be reported
    case TEMPORARY: // should not be reported
    default:
      return false;
  }
}
 

import org.apache.hadoop.hdfs.server.common.HdfsServerConstants.BlockUCState; //导入依赖的package包/类
/**
 * Add a block to the file. Returns a reference to the added block.
 */
BlockInfoContiguous addBlock(String path, INodesInPath inodesInPath,
    Block block, DatanodeStorageInfo[] targets) throws IOException {
  writeLock();
  try {
    final INodeFile fileINode = inodesInPath.getLastINode().asFile();
    Preconditions.checkState(fileINode.isUnderConstruction());

    // check quota limits and updated space consumed
    updateCount(inodesInPath, 0, fileINode.getPreferredBlockSize(),
        fileINode.getBlockReplication(), true);

    // associate new last block for the file
    BlockInfoContiguousUnderConstruction blockInfo =
      new BlockInfoContiguousUnderConstruction(
          block,
          fileINode.getFileReplication(),
          BlockUCState.UNDER_CONSTRUCTION,
          targets);
    getBlockManager().addBlockCollection(blockInfo, fileINode);
    fileINode.addBlock(blockInfo);

    if(NameNode.stateChangeLog.isDebugEnabled()) {
      NameNode.stateChangeLog.debug("DIR* FSDirectory.addBlock: "
          + path + " with " + block
          + " block is added to the in-memory "
          + "file system");
    }
    return blockInfo;
  } finally {
    writeUnlock();
  }
}
 

import org.apache.hadoop.hdfs.server.common.HdfsServerConstants.BlockUCState; //导入依赖的package包/类
/**
 * Faster version of {@link #addStoredBlock},
 * intended for use with initial block report at startup. If not in startup
 * safe mode, will call standard addStoredBlock(). Assumes this method is
 * called "immediately" so there is no need to refresh the storedBlock from
 * blocksMap. Doesn't handle underReplication/overReplication, or worry about
 * pendingReplications or corruptReplicas, because it's in startup safe mode.
 * Doesn't log every block, because there are typically millions of them.
 * 
 * @throws IOException
 */
private void addStoredBlockImmediate(BlockInfoContiguous storedBlock,
    DatanodeStorageInfo storageInfo)
throws IOException {
  assert (storedBlock != null && namesystem.hasWriteLock());
  if (!namesystem.isInStartupSafeMode() 
      || namesystem.isPopulatingReplQueues()) {
    addStoredBlock(storedBlock, storageInfo, null, false);
    return;
  }

  // just add it
  AddBlockResult result = storageInfo.addBlock(storedBlock);

  // Now check for completion of blocks and safe block count
  int numCurrentReplica = countLiveNodes(storedBlock);
  if (storedBlock.getBlockUCState() == BlockUCState.COMMITTED
      && numCurrentReplica >= minReplication) {
    completeBlock(storedBlock.getBlockCollection(), storedBlock, false);
  } else if (storedBlock.isComplete() && result == AddBlockResult.ADDED) {
    // check whether safe replication is reached for the block
    // only complete blocks are counted towards that.
    // In the case that the block just became complete above, completeBlock()
    // handles the safe block count maintenance.
    namesystem.incrementSafeBlockCount(numCurrentReplica);
  }
}
 

import org.apache.hadoop.hdfs.server.common.HdfsServerConstants.BlockUCState; //导入依赖的package包/类
/**
 * Create a block that is currently being constructed.
 */
public BlockInfoContiguousUnderConstruction(Block blk, short replication, BlockUCState state, DatanodeStorageInfo[] targets) {
  super(blk, replication);
  assert getBlockUCState() != BlockUCState.COMPLETE :
    "BlockInfoUnderConstruction cannot be in COMPLETE state";
  this.blockUCState = state;
  setExpectedLocations(targets);
}
 

import org.apache.hadoop.hdfs.server.common.HdfsServerConstants.BlockUCState; //导入依赖的package包/类
/**
 * Commit block's length and generation stamp as reported by the client.
 * Set block state to {@link BlockUCState#COMMITTED}.
 * @param block - contains client reported block length and generation 
 * @throws IOException if block ids are inconsistent.
 */
void commitBlock(Block block) throws IOException {
  if(getBlockId() != block.getBlockId())
    throw new IOException("Trying to commit inconsistent block: id = "
        + block.getBlockId() + ", expected id = " + getBlockId());
  blockUCState = BlockUCState.COMMITTED;
  this.set(getBlockId(), block.getNumBytes(), block.getGenerationStamp());
  // Sort out invalid replicas.
  setGenerationStampAndVerifyReplicas(block.getGenerationStamp());
}
 

import org.apache.hadoop.hdfs.server.common.HdfsServerConstants.BlockUCState; //导入依赖的package包/类
/**
 * Test adding new blocks but without closing the corresponding the file
 */
@Test
public void testAddBlockUC() throws Exception {
  DistributedFileSystem fs = cluster.getFileSystem();
  final Path file1 = new Path("/file1");
  DFSTestUtil.createFile(fs, file1, BLOCKSIZE - 1, REPLICATION, 0L);
  
  FSDataOutputStream out = null;
  try {
    // append files without closing the streams
    out = fs.append(file1);
    String appendContent = "appending-content";
    out.writeBytes(appendContent);
    ((DFSOutputStream) out.getWrappedStream()).hsync(
        EnumSet.of(SyncFlag.UPDATE_LENGTH));
    
    // restart NN
    cluster.restartNameNode(true);
    FSDirectory fsdir = cluster.getNamesystem().getFSDirectory();
    
    INodeFile fileNode = fsdir.getINode4Write(file1.toString()).asFile();
    BlockInfoContiguous[] fileBlocks = fileNode.getBlocks();
    assertEquals(2, fileBlocks.length);
    assertEquals(BLOCKSIZE, fileBlocks[0].getNumBytes());
    assertEquals(BlockUCState.COMPLETE, fileBlocks[0].getBlockUCState());
    assertEquals(appendContent.length() - 1, fileBlocks[1].getNumBytes());
    assertEquals(BlockUCState.UNDER_CONSTRUCTION,
        fileBlocks[1].getBlockUCState());
  } finally {
    if (out != null) {
      out.close();
    }
  }
}
 

import org.apache.hadoop.hdfs.server.common.HdfsServerConstants.BlockUCState; //导入依赖的package包/类
@Override // BlockCollection, the file should be under construction
public void convertLastBlockToUC(BlockInfo lastBlock,
    DatanodeStorageInfo[] locations) throws IOException {
  Preconditions.checkState(isUnderConstruction(),
      "file is no longer under construction");
  if (numBlocks() == 0) {
    throw new IOException("Failed to set last block: File is empty.");
  }
  lastBlock.convertToBlockUnderConstruction(BlockUCState.UNDER_CONSTRUCTION,
      locations);
}
 

import org.apache.hadoop.hdfs.server.common.HdfsServerConstants.BlockUCState; //导入依赖的package包/类
/**
 * Add/Update the under construction feature.
 */
public void convertToBlockUnderConstruction(BlockUCState s,
    DatanodeStorageInfo[] targets) {
  if (isComplete()) {
    uc = new BlockUnderConstructionFeature(this, s, targets,
        this.isStriped());
  } else {
    // the block is already under construction
    uc.setBlockUCState(s);
    uc.setExpectedLocations(this, targets, this.isStriped());
  }
}
 

import org.apache.hadoop.hdfs.server.common.HdfsServerConstants.BlockUCState; //导入依赖的package包/类
/**
 * Convert a specified block of the file to a complete block.
 * @throws IOException if the block does not have at least a minimal number
 * of replicas reported from data-nodes.
 */
private void completeBlock(BlockInfo curBlock, boolean force)
    throws IOException {
  if (curBlock.isComplete()) {
    return;
  }

  int numNodes = curBlock.numNodes();
  if (!force && !hasMinStorage(curBlock, numNodes)) {
    throw new IOException("Cannot complete block: "
        + "block does not satisfy minimal replication requirement.");
  }
  if (!force && curBlock.getBlockUCState() != BlockUCState.COMMITTED) {
    throw new IOException(
        "Cannot complete block: block has not been COMMITTED by the client");
  }

  curBlock.convertToCompleteBlock();
  // Since safe-mode only counts complete blocks, and we now have
  // one more complete block, we need to adjust the total up, and
  // also count it as safe, if we have at least the minimum replica
  // count. (We may not have the minimum replica count yet if this is
  // a "forced" completion when a file is getting closed by an
  // OP_CLOSE edit on the standby).
  bmSafeMode.adjustBlockTotals(0, 1);
  final int minStorage = curBlock.isStriped() ?
      ((BlockInfoStriped) curBlock).getRealDataBlockNum() : minReplication;
  bmSafeMode.incrementSafeBlockCount(Math.min(numNodes, minStorage),
      curBlock);
}
 

import org.apache.hadoop.hdfs.server.common.HdfsServerConstants.BlockUCState; //导入依赖的package包/类
/**
 * Faster version of {@link #addStoredBlock},
 * intended for use with initial block report at startup. If not in startup
 * safe mode, will call standard addStoredBlock(). Assumes this method is
 * called "immediately" so there is no need to refresh the storedBlock from
 * blocksMap. Doesn't handle underReplication/overReplication, or worry about
 * pendingReplications or corruptReplicas, because it's in startup safe mode.
 * Doesn't log every block, because there are typically millions of them.
 * 
 * @throws IOException
 */
private void addStoredBlockImmediate(BlockInfo storedBlock, Block reported,
    DatanodeStorageInfo storageInfo)
throws IOException {
  assert (storedBlock != null && namesystem.hasWriteLock());
  if (!namesystem.isInStartupSafeMode()
      || isPopulatingReplQueues()) {
    addStoredBlock(storedBlock, reported, storageInfo, null, false);
    return;
  }

  // just add it
  AddBlockResult result = storageInfo.addBlock(storedBlock, reported);

  // Now check for completion of blocks and safe block count
  int numCurrentReplica = countLiveNodes(storedBlock);
  if (storedBlock.getBlockUCState() == BlockUCState.COMMITTED
      && hasMinStorage(storedBlock, numCurrentReplica)) {
    completeBlock(storedBlock, false);
  } else if (storedBlock.isComplete() && result == AddBlockResult.ADDED) {
    // check whether safe replication is reached for the block
    // only complete blocks are counted towards that.
    // In the case that the block just became complete above, completeBlock()
    // handles the safe block count maintenance.
    bmSafeMode.incrementSafeBlockCount(numCurrentReplica, storedBlock);
  }
}
 

import org.apache.hadoop.hdfs.server.common.HdfsServerConstants.BlockUCState; //导入依赖的package包/类
public BlockUnderConstructionFeature(Block blk,
    BlockUCState state, DatanodeStorageInfo[] targets, boolean isStriped) {
  assert getBlockUCState() != COMPLETE :
      "BlockUnderConstructionFeature cannot be in COMPLETE state";
  this.blockUCState = state;
  setExpectedLocations(blk, targets, isStriped);
}
 

import org.apache.hadoop.hdfs.server.common.HdfsServerConstants.BlockUCState; //导入依赖的package包/类
/**
 * If the block is committed/completed and its length is less than a full
 * stripe, it returns the the number of actual data blocks.
 * Otherwise it returns the number of data units specified by erasure coding policy.
 */
public short getRealDataBlockNum() {
  if (isComplete() || getBlockUCState() == BlockUCState.COMMITTED) {
    return (short) Math.min(getDataBlockNum(),
        (getNumBytes() - 1) / ecPolicy.getCellSize() + 1);
  } else {
    return getDataBlockNum();
  }
}
 

import org.apache.hadoop.hdfs.server.common.HdfsServerConstants.BlockUCState; //导入依赖的package包/类
/**
 * Test adding new blocks but without closing the corresponding the file
 */
@Test
public void testAddBlockUC() throws Exception {
  DistributedFileSystem fs = cluster.getFileSystem();
  final Path file1 = new Path("/file1");
  DFSTestUtil.createFile(fs, file1, BLOCKSIZE - 1, REPLICATION, 0L);
  
  FSDataOutputStream out = null;
  try {
    // append files without closing the streams
    out = fs.append(file1);
    String appendContent = "appending-content";
    out.writeBytes(appendContent);
    ((DFSOutputStream) out.getWrappedStream()).hsync(
        EnumSet.of(SyncFlag.UPDATE_LENGTH));
    
    // restart NN
    cluster.restartNameNode(true);
    FSDirectory fsdir = cluster.getNamesystem().getFSDirectory();
    
    INodeFile fileNode = fsdir.getINode4Write(file1.toString()).asFile();
    BlockInfo[] fileBlocks = fileNode.getBlocks();
    assertEquals(2, fileBlocks.length);
    assertEquals(BLOCKSIZE, fileBlocks[0].getNumBytes());
    assertEquals(BlockUCState.COMPLETE, fileBlocks[0].getBlockUCState());
    assertEquals(appendContent.length() - 1, fileBlocks[1].getNumBytes());
    assertEquals(BlockUCState.UNDER_CONSTRUCTION,
        fileBlocks[1].getBlockUCState());
  } finally {
    if (out != null) {
      out.close();
    }
  }
}
 

import org.apache.hadoop.hdfs.server.common.HdfsServerConstants.BlockUCState; //导入依赖的package包/类
/**
 * Add a block to the file. Returns a reference to the added block.
 */
BlockInfo addBlock(String path, INodesInPath inodesInPath, Block block,
    DatanodeStorageInfo[] targets) throws IOException {
  writeLock();
  try {
    final INodeFile fileINode = inodesInPath.getLastINode().asFile();
    Preconditions.checkState(fileINode.isUnderConstruction());

    // check quota limits and updated space consumed
    updateCount(inodesInPath, 0, fileINode.getBlockDiskspace(), true);

    // associate new last block for the file
    BlockInfoUnderConstruction blockInfo =
      new BlockInfoUnderConstruction(
          block,
          fileINode.getFileReplication(),
          BlockUCState.UNDER_CONSTRUCTION,
          targets);
    getBlockManager().addBlockCollection(blockInfo, fileINode);
    fileINode.addBlock(blockInfo);

    if(NameNode.stateChangeLog.isDebugEnabled()) {
      NameNode.stateChangeLog.debug("DIR* FSDirectory.addBlock: "
          + path + " with " + block
          + " block is added to the in-memory "
          + "file system");
    }
    return blockInfo;
  } finally {
    writeUnlock();
  }
}