import org.apache.hadoop.hdfs.server.common.HdfsServerConstants.ReplicaState; //导入依赖的package包/类
public void add(Replica replica) {
  try {
    // zig-zag to reduce size of legacy blocks
    cos.writeSInt64NoTag(replica.getBlockId());
    cos.writeRawVarint64(replica.getBytesOnDisk());
    cos.writeRawVarint64(replica.getGenerationStamp());
    ReplicaState state = replica.getState();
    // although state is not a 64-bit value, using a long varint to
    // allow for future use of the upper bits
    cos.writeRawVarint64(state.getValue());
    if (state == ReplicaState.FINALIZED) {
      numFinalized++;
    }
    numBlocks++;
  } catch (IOException ioe) {
    // shouldn't happen, ByteString.Output doesn't throw IOE
    throw new IllegalStateException(ioe);
  }
}
 

import org.apache.hadoop.hdfs.server.common.HdfsServerConstants.ReplicaState; //导入依赖的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.ReplicaState; //导入依赖的package包/类
void addReplicaIfNotPresent(DatanodeStorageInfo storage,
                   Block block,
                   ReplicaState rState) {
  Iterator<ReplicaUnderConstruction> it = replicas.iterator();
  while (it.hasNext()) {
    ReplicaUnderConstruction r = it.next();
    DatanodeStorageInfo expectedLocation = r.getExpectedStorageLocation();
    if(expectedLocation == storage) {
      // Record the gen stamp from the report
      r.setGenerationStamp(block.getGenerationStamp());
      return;
    } else if (expectedLocation != null &&
               expectedLocation.getDatanodeDescriptor() ==
                   storage.getDatanodeDescriptor()) {

      // The Datanode reported that the block is on a different storage
      // than the one chosen by BlockPlacementPolicy. This can occur as
      // we allow Datanodes to choose the target storage. Update our
      // state by removing the stale entry and adding a new one.
      it.remove();
      break;
    }
  }
  replicas.add(new ReplicaUnderConstruction(block, storage, rState));
}
 

import org.apache.hadoop.hdfs.server.common.HdfsServerConstants.ReplicaState; //导入依赖的package包/类
/**
 * Remove the temporary block file (if any)
 */
@Override // FsDatasetSpi
public synchronized void unfinalizeBlock(ExtendedBlock b) throws IOException {
  ReplicaInfo replicaInfo = volumeMap.get(b.getBlockPoolId(), 
      b.getLocalBlock());
  if (replicaInfo != null && replicaInfo.getState() == ReplicaState.TEMPORARY) {
    // remove from volumeMap
    volumeMap.remove(b.getBlockPoolId(), b.getLocalBlock());
    
    // delete the on-disk temp file
    if (delBlockFromDisk(replicaInfo.getBlockFile(), 
        replicaInfo.getMetaFile(), b.getLocalBlock())) {
      LOG.warn("Block " + b + " unfinalized and removed. " );
    }
    if (replicaInfo.getVolume().isTransientStorage()) {
      ramDiskReplicaTracker.discardReplica(b.getBlockPoolId(), b.getBlockId(), true);
    }
  }
}
 

import org.apache.hadoop.hdfs.server.common.HdfsServerConstants.ReplicaState; //导入依赖的package包/类
/**
 * Check if a block is valid.
 *
 * @param b           The block to check.
 * @param minLength   The minimum length that the block must have.  May be 0.
 * @param state       If this is null, it is ignored.  If it is non-null, we
 *                        will check that the replica has this state.
 *
 * @throws ReplicaNotFoundException          If the replica is not found 
 *
 * @throws UnexpectedReplicaStateException   If the replica is not in the 
 *                                             expected state.
 * @throws FileNotFoundException             If the block file is not found or there
 *                                              was an error locating it.
 * @throws EOFException                      If the replica length is too short.
 * 
 * @throws IOException                       May be thrown from the methods called. 
 */
public void checkBlock(ExtendedBlock b, long minLength, ReplicaState state)
    throws ReplicaNotFoundException, UnexpectedReplicaStateException,
    FileNotFoundException, EOFException, IOException {
  final ReplicaInfo replicaInfo = volumeMap.get(b.getBlockPoolId(), 
      b.getLocalBlock());
  if (replicaInfo == null) {
    throw new ReplicaNotFoundException(b);
  }
  if (replicaInfo.getState() != state) {
    throw new UnexpectedReplicaStateException(b,state);
  }
  if (!replicaInfo.getBlockFile().exists()) {
    throw new FileNotFoundException(replicaInfo.getBlockFile().getPath());
  }
  long onDiskLength = getLength(b);
  if (onDiskLength < minLength) {
    throw new EOFException(b + "'s on-disk length " + onDiskLength
        + " is shorter than minLength " + minLength);
  }
}
 

import org.apache.hadoop.hdfs.server.common.HdfsServerConstants.ReplicaState; //导入依赖的package包/类
@Test
public void testMix() {
  BlockListAsLongs blocks = checkReport(
      new FinalizedReplica(b1, null, null),
      new FinalizedReplica(b2, null, null),
      new ReplicaBeingWritten(b3, null, null, null),
      new ReplicaWaitingToBeRecovered(b4, null, null));
  assertArrayEquals(
      new long[] {
          2, 2,
          1, 11, 111,
          2, 22, 222,
          -1, -1, -1,
          3, 33, 333, ReplicaState.RBW.getValue(),
          4, 44, 444, ReplicaState.RWR.getValue() },
      blocks.getBlockListAsLongs());
}
 

import org.apache.hadoop.hdfs.server.common.HdfsServerConstants.ReplicaState; //导入依赖的package包/类
@Test
public void testQueues() {
  DatanodeDescriptor fakeDN = DFSTestUtil.getLocalDatanodeDescriptor();
  DatanodeStorage storage = new DatanodeStorage("STORAGE_ID");
  DatanodeStorageInfo storageInfo = new DatanodeStorageInfo(fakeDN, storage);
  msgs.enqueueReportedBlock(storageInfo, block1Gs1, ReplicaState.FINALIZED);
  msgs.enqueueReportedBlock(storageInfo, block1Gs2, ReplicaState.FINALIZED);

  assertEquals(2, msgs.count());
  
  // Nothing queued yet for block 2
  assertNull(msgs.takeBlockQueue(block2Gs1));
  assertEquals(2, msgs.count());
  
  Queue<ReportedBlockInfo> q =
    msgs.takeBlockQueue(block1Gs2DifferentInstance);
  assertEquals(
      "ReportedBlockInfo [block=blk_1_1, dn=127.0.0.1:50010, reportedState=FINALIZED]," +
      "ReportedBlockInfo [block=blk_1_2, dn=127.0.0.1:50010, reportedState=FINALIZED]",
      Joiner.on(",").join(q));
  assertEquals(0, msgs.count());
  
  // Should be null if we pull again
  assertNull(msgs.takeBlockQueue(block1Gs1));
  assertEquals(0, msgs.count());
}
 

import org.apache.hadoop.hdfs.server.common.HdfsServerConstants.ReplicaState; //导入依赖的package包/类
/**
 * BlockRecovery_02.11.
 * Two replicas are RBW.
 * @throws IOException in case of an error
 */
@Test
public void testRBWReplicas() throws IOException {
  if(LOG.isDebugEnabled()) {
    LOG.debug("Running " + GenericTestUtils.getMethodName());
  }
  ReplicaRecoveryInfo replica1 = new ReplicaRecoveryInfo(BLOCK_ID, 
      REPLICA_LEN1, GEN_STAMP-1, ReplicaState.RBW);
  ReplicaRecoveryInfo replica2 = new ReplicaRecoveryInfo(BLOCK_ID, 
      REPLICA_LEN2, GEN_STAMP-2, ReplicaState.RBW);

  InterDatanodeProtocol dn1 = mock(InterDatanodeProtocol.class);
  InterDatanodeProtocol dn2 = mock(InterDatanodeProtocol.class);

  long minLen = Math.min(REPLICA_LEN1, REPLICA_LEN2);
  testSyncReplicas(replica1, replica2, dn1, dn2, minLen);
  verify(dn1).updateReplicaUnderRecovery(block, RECOVERY_ID, BLOCK_ID, minLen);
  verify(dn2).updateReplicaUnderRecovery(block, RECOVERY_ID, BLOCK_ID, minLen);
}
 

import org.apache.hadoop.hdfs.server.common.HdfsServerConstants.ReplicaState; //导入依赖的package包/类
/**
 * BlockRecovery_02.12.
 * One replica is RBW and another is RWR. 
 * @throws IOException in case of an error
 */
@Test
public void testRBW_RWRReplicas() throws IOException {
  if(LOG.isDebugEnabled()) {
    LOG.debug("Running " + GenericTestUtils.getMethodName());
  }
  ReplicaRecoveryInfo replica1 = new ReplicaRecoveryInfo(BLOCK_ID, 
      REPLICA_LEN1, GEN_STAMP-1, ReplicaState.RBW);
  ReplicaRecoveryInfo replica2 = new ReplicaRecoveryInfo(BLOCK_ID, 
      REPLICA_LEN1, GEN_STAMP-2, ReplicaState.RWR);

  InterDatanodeProtocol dn1 = mock(InterDatanodeProtocol.class);
  InterDatanodeProtocol dn2 = mock(InterDatanodeProtocol.class);

  testSyncReplicas(replica1, replica2, dn1, dn2, REPLICA_LEN1);
  verify(dn1).updateReplicaUnderRecovery(block, RECOVERY_ID, BLOCK_ID,
      REPLICA_LEN1);
  verify(dn2, never()).updateReplicaUnderRecovery(
      block, RECOVERY_ID, BLOCK_ID, REPLICA_LEN1);
}
 

import org.apache.hadoop.hdfs.server.common.HdfsServerConstants.ReplicaState; //导入依赖的package包/类
/**
 * BlockRecovery_02.13. 
 * Two replicas are RWR.
 * @throws IOException in case of an error
 */
@Test
public void testRWRReplicas() throws IOException {
  if(LOG.isDebugEnabled()) {
    LOG.debug("Running " + GenericTestUtils.getMethodName());
  }
  ReplicaRecoveryInfo replica1 = new ReplicaRecoveryInfo(BLOCK_ID, 
      REPLICA_LEN1, GEN_STAMP-1, ReplicaState.RWR);
  ReplicaRecoveryInfo replica2 = new ReplicaRecoveryInfo(BLOCK_ID, 
      REPLICA_LEN2, GEN_STAMP-2, ReplicaState.RWR);

  InterDatanodeProtocol dn1 = mock(InterDatanodeProtocol.class);
  InterDatanodeProtocol dn2 = mock(InterDatanodeProtocol.class);

  long minLen = Math.min(REPLICA_LEN1, REPLICA_LEN2);
  testSyncReplicas(replica1, replica2, dn1, dn2, minLen);
  
  verify(dn1).updateReplicaUnderRecovery(block, RECOVERY_ID, BLOCK_ID, minLen);
  verify(dn2).updateReplicaUnderRecovery(block, RECOVERY_ID, BLOCK_ID, minLen);
}
 

import org.apache.hadoop.hdfs.server.common.HdfsServerConstants.ReplicaState; //导入依赖的package包/类
/**
 * BlockRecoveryFI_07. max replica length from all DNs is zero.
 *
 * @throws IOException in case of an error
 */
@Test
public void testZeroLenReplicas() throws IOException, InterruptedException {
  if(LOG.isDebugEnabled()) {
    LOG.debug("Running " + GenericTestUtils.getMethodName());
  }
  DataNode spyDN = spy(dn);
  doReturn(new ReplicaRecoveryInfo(block.getBlockId(), 0,
      block.getGenerationStamp(), ReplicaState.FINALIZED)).when(spyDN).
      initReplicaRecovery(any(RecoveringBlock.class));
  Daemon d = spyDN.recoverBlocks("fake NN", initRecoveringBlocks());
  d.join();
  DatanodeProtocol dnP = dn.getActiveNamenodeForBP(POOL_ID);
  verify(dnP).commitBlockSynchronization(
      block, RECOVERY_ID, 0, true, true, DatanodeID.EMPTY_ARRAY, null);
}
 

import org.apache.hadoop.hdfs.server.common.HdfsServerConstants.ReplicaState; //导入依赖的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.ReplicaState; //导入依赖的package包/类
/**
 * Remove the temporary block file (if any)
 */
@Override // FsDatasetSpi
public synchronized void unfinalizeBlock(ExtendedBlock b) throws IOException {
  ReplicaInfo replicaInfo = volumeMap.get(b.getBlockPoolId(), 
      b.getLocalBlock());
  if (replicaInfo != null && replicaInfo.getState() == ReplicaState.TEMPORARY) {
    // remove from volumeMap
    volumeMap.remove(b.getBlockPoolId(), b.getLocalBlock());

    // delete the on-disk temp file
    if (delBlockFromDisk(replicaInfo.getBlockFile(), 
        replicaInfo.getMetaFile(), b.getLocalBlock())) {
      LOG.warn("Block " + b + " unfinalized and removed. " );
    }
    if (replicaInfo.getVolume().isTransientStorage()) {
      ramDiskReplicaTracker.discardReplica(b.getBlockPoolId(), b.getBlockId(), true);
    }
  }
}
 

import org.apache.hadoop.hdfs.server.common.HdfsServerConstants.ReplicaState; //导入依赖的package包/类
/**
 * BlockRecovery_02.12.
 * One replica is RBW and another is RWR. 
 * @throws IOException in case of an error
 */
@Test
public void testRBW_RWRReplicas() throws IOException {
  if(LOG.isDebugEnabled()) {
    LOG.debug("Running " + GenericTestUtils.getMethodName());
  }
  ReplicaRecoveryInfo replica1 = new ReplicaRecoveryInfo(BLOCK_ID, 
      REPLICA_LEN1, GEN_STAMP-1, ReplicaState.RBW);
  ReplicaRecoveryInfo replica2 = new ReplicaRecoveryInfo(BLOCK_ID, 
      REPLICA_LEN1, GEN_STAMP-2, ReplicaState.RWR);

  InterDatanodeProtocol dn1 = mock(InterDatanodeProtocol.class);
  InterDatanodeProtocol dn2 = mock(InterDatanodeProtocol.class);

  testSyncReplicas(replica1, replica2, dn1, dn2, REPLICA_LEN1);
  verify(dn1).updateReplicaUnderRecovery(block, RECOVERY_ID, BLOCK_ID, REPLICA_LEN1);
  verify(dn2, never()).updateReplicaUnderRecovery(
      block, RECOVERY_ID, BLOCK_ID, REPLICA_LEN1);
}
 

import org.apache.hadoop.hdfs.server.common.HdfsServerConstants.ReplicaState; //导入依赖的package包/类
/**
 * BlockRecoveryFI_07. max replica length from all DNs is zero.
 *
 * @throws IOException in case of an error
 */
@Test
public void testZeroLenReplicas() throws IOException, InterruptedException {
  if(LOG.isDebugEnabled()) {
    LOG.debug("Running " + GenericTestUtils.getMethodName());
  }
  doReturn(new ReplicaRecoveryInfo(block.getBlockId(), 0,
      block.getGenerationStamp(), ReplicaState.FINALIZED)).when(spyDN).
      initReplicaRecovery(any(RecoveringBlock.class));

  for(RecoveringBlock rBlock: initRecoveringBlocks()){
    BlockRecoveryWorker.RecoveryTaskContiguous RecoveryTaskContiguous =
        recoveryWorker.new RecoveryTaskContiguous(rBlock);
    BlockRecoveryWorker.RecoveryTaskContiguous spyTask
        = spy(RecoveryTaskContiguous);
    spyTask.recover();
  }
  DatanodeProtocol dnP = recoveryWorker.getActiveNamenodeForBP(POOL_ID);
  verify(dnP).commitBlockSynchronization(
      block, RECOVERY_ID, 0, true, true, DatanodeID.EMPTY_ARRAY, null);
}
 

import org.apache.hadoop.hdfs.server.common.HdfsServerConstants.ReplicaState; //导入依赖的package包/类
@Override
public Iterator<BlockReportReplica> iterator() {
  return new Iterator<BlockReportReplica>() {
    final BlockReportReplica block = new BlockReportReplica();
    final CodedInputStream cis = buffer.newCodedInput();
    private int currentBlockIndex = 0;

    @Override
    public boolean hasNext() {
      return currentBlockIndex < numBlocks;
    }

    @Override
    public BlockReportReplica next() {
      currentBlockIndex++;
      try {
        // zig-zag to reduce size of legacy blocks and mask off bits
        // we don't (yet) understand
        block.setBlockId(cis.readSInt64());
        block.setNumBytes(cis.readRawVarint64() & NUM_BYTES_MASK);
        block.setGenerationStamp(cis.readRawVarint64());
        long state = cis.readRawVarint64() & REPLICA_STATE_MASK;
        block.setState(ReplicaState.getState((int)state));
      } catch (IOException e) {
        throw new IllegalStateException(e);
      }
      return block;
    }

    @Override
    public void remove() {
      throw new UnsupportedOperationException();
    }
  };
}
 

import org.apache.hadoop.hdfs.server.common.HdfsServerConstants.ReplicaState; //导入依赖的package包/类
public BlockReportReplica(Block block) {
  super(block);
  if (block instanceof BlockReportReplica) {
    this.state = ((BlockReportReplica)block).getState();
  } else {
    this.state = ReplicaState.FINALIZED;
  }
}
 

import org.apache.hadoop.hdfs.server.common.HdfsServerConstants.ReplicaState; //导入依赖的package包/类
/**
 * Queue the given reported block for later processing in the
 * standby node. @see PendingDataNodeMessages.
 * @param reason a textual reason to report in the debug logs
 */
private void queueReportedBlock(DatanodeStorageInfo storageInfo, Block block,
    ReplicaState reportedState, String reason) {
  assert shouldPostponeBlocksFromFuture;
  
  if (LOG.isDebugEnabled()) {
    LOG.debug("Queueing reported block " + block +
        " in state " + reportedState + 
        " from datanode " + storageInfo.getDatanodeDescriptor() +
        " for later processing because " + reason + ".");
  }
  pendingDNMessages.enqueueReportedBlock(storageInfo, block, reportedState);
}
 

import org.apache.hadoop.hdfs.server.common.HdfsServerConstants.ReplicaState; //导入依赖的package包/类
void addStoredBlockUnderConstruction(StatefulBlockInfo ucBlock,
    DatanodeStorageInfo storageInfo) throws IOException {
  BlockInfoContiguousUnderConstruction block = ucBlock.storedBlock;
  block.addReplicaIfNotPresent(
      storageInfo, ucBlock.reportedBlock, ucBlock.reportedState);

  if (ucBlock.reportedState == ReplicaState.FINALIZED &&
      !block.findDatanode(storageInfo.getDatanodeDescriptor())) {
    addStoredBlock(block, storageInfo, null, true);
  }
}
 

import org.apache.hadoop.hdfs.server.common.HdfsServerConstants.ReplicaState; //导入依赖的package包/类
/**
 * The given node is reporting that it received a certain block.
 */
@VisibleForTesting
void addBlock(DatanodeStorageInfo storageInfo, Block block, String delHint)
    throws IOException {
  DatanodeDescriptor node = storageInfo.getDatanodeDescriptor();
  // Decrement number of blocks scheduled to this datanode.
  // for a retry request (of DatanodeProtocol#blockReceivedAndDeleted with 
  // RECEIVED_BLOCK), we currently also decrease the approximate number. 
  node.decrementBlocksScheduled(storageInfo.getStorageType());

  // get the deletion hint node
  DatanodeDescriptor delHintNode = null;
  if (delHint != null && delHint.length() != 0) {
    delHintNode = datanodeManager.getDatanode(delHint);
    if (delHintNode == null) {
      blockLog.warn("BLOCK* blockReceived: {} is expected to be removed " +
          "from an unrecorded node {}", block, delHint);
    }
  }

  //
  // Modify the blocks->datanode map and node's map.
  //
  pendingReplications.decrement(block, node);
  processAndHandleReportedBlock(storageInfo, block, ReplicaState.FINALIZED,
      delHintNode);
}