import org.apache.hadoop.hdfs.protocol.RecoveryInProgressException; //导入依赖的package包/类
/** Try openning a file for append. */
private static FSDataOutputStream append(FileSystem fs, Path p) throws Exception {
  for(int i = 0; i < 10; i++) {
    try {
      return fs.append(p);
    } catch(RemoteException re) {
      if (re.getClassName().equals(RecoveryInProgressException.class.getName())) {
        AppendTestUtil.LOG.info("Will sleep and retry, i=" + i +", p="+p, re);
        Thread.sleep(1000);
      }
      else
        throw re;
    }
  }
  throw new IOException("Cannot append to " + p);
}
 

import org.apache.hadoop.hdfs.protocol.RecoveryInProgressException; //导入依赖的package包/类
/**
 * BlockRecoveryFI_05. One DN throws RecoveryInProgressException.
 *
 * @throws IOException
 *           in case of an error
 */
@Test
public void testRecoveryInProgressException()
  throws IOException, InterruptedException {
  if(LOG.isDebugEnabled()) {
    LOG.debug("Running " + GenericTestUtils.getMethodName());
  }
  doThrow(new RecoveryInProgressException("Replica recovery is in progress")).
     when(spyDN).initReplicaRecovery(any(RecoveringBlock.class));

  for(RecoveringBlock rBlock: initRecoveringBlocks()){
    BlockRecoveryWorker.RecoveryTaskContiguous RecoveryTaskContiguous =
        recoveryWorker.new RecoveryTaskContiguous(rBlock);
    BlockRecoveryWorker.RecoveryTaskContiguous spyTask
        = spy(RecoveryTaskContiguous);
    spyTask.recover();
    verify(spyTask, never()).syncBlock(anyListOf(BlockRecord.class));
  }
}
 

import org.apache.hadoop.hdfs.protocol.RecoveryInProgressException; //导入依赖的package包/类
/**
 * Try openning a file for append.
 */
private static FSDataOutputStream append(FileSystem fs, Path p)
    throws Exception {
  for (int i = 0; i < 10; i++) {
    try {
      return fs.append(p);
    } catch (RemoteException re) {
      if (re.getClassName().equals(
          RecoveryInProgressException.NonAbortingRecoveryInProgressException.class
              .getName())) {
        AppendTestUtil.LOG
            .info("Will sleep and retry, i=" + i + ", p=" + p, re);
        Thread.sleep(1000);
      } else {
        throw re;
      }
    }
  }
  throw new IOException("Cannot append to " + p);
}
 

import org.apache.hadoop.hdfs.protocol.RecoveryInProgressException; //导入依赖的package包/类
/**
 * BlockRecoveryFI_05. One DN throws RecoveryInProgressException.
 *
 * @throws IOException
 *     in case of an error
 */
@Test
public void testRecoveryInProgressException()
    throws IOException, InterruptedException {
  if (LOG.isDebugEnabled()) {
    LOG.debug("Running " + GenericTestUtils.getMethodName());
  }
  try {
    DataNode spyDN = spy(dn);
    doThrow(
        new RecoveryInProgressException("Replica recovery is in progress")).
        when(spyDN).initReplicaRecovery(any(RecoveringBlock.class));
    Daemon d = spyDN.recoverBlocks("fake NN", initRecoveringBlocks());
    d.join();
    verify(spyDN, never())
        .syncBlock(any(RecoveringBlock.class), anyListOf(BlockRecord.class));
  } catch (Exception e) {
    e.printStackTrace();
  }

}
 

import org.apache.hadoop.hdfs.protocol.RecoveryInProgressException; //导入依赖的package包/类
/**
 * BlockRecoveryFI_05. One DN throws RecoveryInProgressException.
 *
 * @throws IOException
 *           in case of an error
 */
@Test
public void testRecoveryInProgressException()
  throws IOException, InterruptedException {
  if(LOG.isDebugEnabled()) {
    LOG.debug("Running " + GenericTestUtils.getMethodName());
  }
  DataNode spyDN = spy(dn);
  doThrow(new RecoveryInProgressException("Replica recovery is in progress")).
     when(spyDN).initReplicaRecovery(any(RecoveringBlock.class));
  Daemon d = spyDN.recoverBlocks("fake NN", initRecoveringBlocks());
  d.join();
  verify(spyDN, never()).syncBlock(
      any(RecoveringBlock.class), anyListOf(BlockRecord.class));
}
 

import org.apache.hadoop.hdfs.protocol.RecoveryInProgressException; //导入依赖的package包/类
@Override
protected final boolean shouldAbort(Exception e) {
  if (e instanceof RecoveryInProgressException.NonAbortingRecoveryInProgressException) {
    return false;
  }
  return true;
}
 

import org.apache.hadoop.hdfs.protocol.RecoveryInProgressException; //导入依赖的package包/类
/**
 * BlockRecoveryFI_05. One DN throws RecoveryInProgressException.
 *
 * @throws IOException
 *           in case of an error
 */
@Test
public void testRecoveryInProgressException()
  throws IOException, InterruptedException {
  if(LOG.isDebugEnabled()) {
    LOG.debug("Running " + GenericTestUtils.getMethodName());
  }
  DataNode spyDN = spy(dn);
  doThrow(new RecoveryInProgressException("Replica recovery is in progress")).
     when(spyDN).initReplicaRecovery(any(RecoveringBlock.class));
  Daemon d = spyDN.recoverBlocks(initRecoveringBlocks());
  d.join();
  verify(spyDN, never()).syncBlock(
      any(RecoveringBlock.class), anyListOf(BlockRecord.class));
}
 

import org.apache.hadoop.hdfs.protocol.RecoveryInProgressException; //导入依赖的package包/类
/** Recover a block */
private void recoverBlock(RecoveringBlock rBlock) throws IOException {
  ExtendedBlock block = rBlock.getBlock();
  String blookPoolId = block.getBlockPoolId();
  DatanodeID[] datanodeids = rBlock.getLocations();
  List<BlockRecord> syncList = new ArrayList<BlockRecord>(datanodeids.length);
  int errorCount = 0;

  //check generation stamps
  for(DatanodeID id : datanodeids) {
    try {
      BPOfferService bpos = blockPoolManager.get(blookPoolId);
      DatanodeRegistration bpReg = bpos.bpRegistration;
      InterDatanodeProtocol datanode = bpReg.equals(id)?
          this: DataNode.createInterDataNodeProtocolProxy(id, getConf(),
              dnConf.socketTimeout, dnConf.connectToDnViaHostname);
      ReplicaRecoveryInfo info = callInitReplicaRecovery(datanode, rBlock);
      if (info != null &&
          info.getGenerationStamp() >= block.getGenerationStamp() &&
          info.getNumBytes() > 0) {
        syncList.add(new BlockRecord(id, datanode, info));
      }
    } catch (RecoveryInProgressException ripE) {
      InterDatanodeProtocol.LOG.warn(
          "Recovery for replica " + block + " on data-node " + id
          + " is already in progress. Recovery id = "
          + rBlock.getNewGenerationStamp() + " is aborted.", ripE);
      return;
    } catch (IOException e) {
      ++errorCount;
      InterDatanodeProtocol.LOG.warn(
          "Failed to obtain replica info for block (=" + block 
          + ") from datanode (=" + id + ")", e);
    }
  }

  if (errorCount == datanodeids.length) {
    throw new IOException("All datanodes failed: block=" + block
        + ", datanodeids=" + Arrays.asList(datanodeids));
  }

  syncBlock(rBlock, syncList);
}
 

import org.apache.hadoop.hdfs.protocol.RecoveryInProgressException; //导入依赖的package包/类
protected void recover() throws IOException {
  List<BlockRecord> syncList = new ArrayList<>(locs.length);
  int errorCount = 0;
  int candidateReplicaCnt = 0;

  // Check generation stamps, replica size and state. Replica must satisfy
  // the following criteria to be included in syncList for recovery:
  // - Valid generation stamp
  // - Non-zero length
  // - Original state is RWR or better
  for(DatanodeID id : locs) {
    try {
      DatanodeID bpReg = new DatanodeID(
          datanode.getBPOfferService(bpid).bpRegistration);
      InterDatanodeProtocol proxyDN = bpReg.equals(id)?
          datanode: DataNode.createInterDataNodeProtocolProxy(id, conf,
          dnConf.socketTimeout, dnConf.connectToDnViaHostname);
      ReplicaRecoveryInfo info = callInitReplicaRecovery(proxyDN, rBlock);
      if (info != null &&
          info.getGenerationStamp() >= block.getGenerationStamp() &&
          info.getNumBytes() > 0) {
        // Count the number of candidate replicas received.
        ++candidateReplicaCnt;
        if (info.getOriginalReplicaState().getValue() <=
            ReplicaState.RWR.getValue()) {
          syncList.add(new BlockRecord(id, proxyDN, info));
        } else {
          if (LOG.isDebugEnabled()) {
            LOG.debug("Block recovery: Ignored replica with invalid " +
                "original state: " + info + " from DataNode: " + id);
          }
        }
      } else {
        if (LOG.isDebugEnabled()) {
          if (info == null) {
            LOG.debug("Block recovery: DataNode: " + id + " does not have "
                + "replica for block: " + block);
          } else {
            LOG.debug("Block recovery: Ignored replica with invalid "
                + "generation stamp or length: " + info + " from " +
                "DataNode: " + id);
          }
        }
      }
    } catch (RecoveryInProgressException ripE) {
      InterDatanodeProtocol.LOG.warn(
          "Recovery for replica " + block + " on data-node " + id
              + " is already in progress. Recovery id = "
              + rBlock.getNewGenerationStamp() + " is aborted.", ripE);
      return;
    } catch (IOException e) {
      ++errorCount;
      InterDatanodeProtocol.LOG.warn(
          "Failed to obtain replica info for block (=" + block
              + ") from datanode (=" + id + ")", e);
    }
  }

  if (errorCount == locs.length) {
    throw new IOException("All datanodes failed: block=" + block
        + ", datanodeids=" + Arrays.asList(locs));
  }

  // None of the replicas reported by DataNodes has the required original
  // state, report the error.
  if (candidateReplicaCnt > 0 && syncList.isEmpty()) {
    throw new IOException("Found " + candidateReplicaCnt +
        " replica(s) for block " + block + " but none is in " +
        ReplicaState.RWR.name() + " or better state. datanodeids=" +
        Arrays.asList(locs));
  }

  syncBlock(syncList);
}
 

import org.apache.hadoop.hdfs.protocol.RecoveryInProgressException; //导入依赖的package包/类
/**
 * Recover a block
 */
private void recoverBlock(RecoveringBlock rBlock) throws IOException {
  ExtendedBlock block = rBlock.getBlock();
  String blookPoolId = block.getBlockPoolId();
  DatanodeID[] datanodeids = rBlock.getLocations();
  List<BlockRecord> syncList = new ArrayList<>(datanodeids.length);
  int errorCount = 0;

  //check generation stamps
  for (DatanodeID id : datanodeids) {
    try {
      BPOfferService bpos = blockPoolManager.get(blookPoolId);
      DatanodeRegistration bpReg = bpos.bpRegistration;
      InterDatanodeProtocol datanode = bpReg.equals(id) ? this : DataNode
          .createInterDataNodeProtocolProxy(id, getConf(),
              dnConf.socketTimeout, dnConf.connectToDnViaHostname);
      ReplicaRecoveryInfo info = callInitReplicaRecovery(datanode, rBlock);
      if (info != null &&
          info.getGenerationStamp() >= block.getGenerationStamp() &&
          info.getNumBytes() > 0) {
        syncList.add(new BlockRecord(id, datanode, info));
      }
    } catch (RecoveryInProgressException ripE) {
      InterDatanodeProtocol.LOG.warn(
          "Recovery for replica " + block + " on data-node " + id +
              " is already in progress. Recovery id = " +
              rBlock.getNewGenerationStamp() + " is aborted.", ripE);
      return;
    } catch (IOException e) {
      ++errorCount;
      InterDatanodeProtocol.LOG.warn(
          "Failed to obtain replica info for block (=" + block +
              ") from datanode (=" + id + ")", e);
    }
  }

  if (errorCount == datanodeids.length) {
    throw new IOException(
        "All datanodes failed: block=" + block + ", datanodeids=" +
            Arrays.asList(datanodeids));
  }

  syncBlock(rBlock, syncList);
}
 

import org.apache.hadoop.hdfs.protocol.RecoveryInProgressException; //导入依赖的package包/类
/** static version of {@link #initReplicaRecovery(Block, long)}. */
static ReplicaRecoveryInfo initReplicaRecovery(
    ReplicasMap map, Block block, long recoveryId) throws IOException {
  final ReplicaInfo replica = map.get(block.getBlockId());
  DataNode.LOG.info("initReplicaRecovery: block=" + block
      + ", recoveryId=" + recoveryId
      + ", replica=" + replica);

  //check replica
  if (replica == null) {
    return null;
  }

  //stop writer if there is any
  if (replica instanceof ReplicaInPipeline) {
    final ReplicaInPipeline rip = (ReplicaInPipeline)replica;
    rip.stopWriter();

    //check replica bytes on disk.
    if (rip.getBytesOnDisk() < rip.getVisibleLength()) {
      throw new IOException("THIS IS NOT SUPPOSED TO HAPPEN:"
          + " getBytesOnDisk() < getVisibleLength(), rip=" + rip);
    }

    //check the replica's files
    checkReplicaFiles(rip);
  }

  //check generation stamp
  if (replica.getGenerationStamp() < block.getGenerationStamp()) {
    throw new IOException(
        "replica.getGenerationStamp() < block.getGenerationStamp(), block="
        + block + ", replica=" + replica);
  }

  //check recovery id
  if (replica.getGenerationStamp() >= recoveryId) {
    throw new IOException("THIS IS NOT SUPPOSED TO HAPPEN:"
        + " replica.getGenerationStamp() >= recoveryId = " + recoveryId
        + ", block=" + block + ", replica=" + replica);
  }

  //check RUR
  final ReplicaUnderRecovery rur;
  if (replica.getState() == ReplicaState.RUR) {
    rur = (ReplicaUnderRecovery)replica;
    if (rur.getRecoveryID() >= recoveryId) {
      throw new RecoveryInProgressException(
          "rur.getRecoveryID() >= recoveryId = " + recoveryId
          + ", block=" + block + ", rur=" + rur);
    }
    final long oldRecoveryID = rur.getRecoveryID();
    rur.setRecoveryID(recoveryId);
    DataNode.LOG.info("initReplicaRecovery: update recovery id for " + block
        + " from " + oldRecoveryID + " to " + recoveryId);
  }
  else {
    rur = new ReplicaUnderRecovery(replica, recoveryId);
    map.add(rur);
    DataNode.LOG.info("initReplicaRecovery: changing replica state for "
        + block + " from " + replica.getState()
        + " to " + rur.getState());
  }
  return rur.createInfo();
}
 

import org.apache.hadoop.hdfs.protocol.RecoveryInProgressException; //导入依赖的package包/类
/** Recover a block */
private void recoverBlock(RecoveringBlock rBlock) throws IOException {
  Block block = rBlock.getBlock();
  DatanodeInfo[] targets = rBlock.getLocations();
  DatanodeID[] datanodeids = (DatanodeID[])targets;
  List<BlockRecord> syncList = new ArrayList<BlockRecord>(datanodeids.length);
  int errorCount = 0;

  //check generation stamps
  for(DatanodeID id : datanodeids) {
    try {
      InterDatanodeProtocol datanode = dnRegistration.equals(id)?
          this: DataNode.createInterDataNodeProtocolProxy(id, getConf(),
              socketTimeout);
      ReplicaRecoveryInfo info = callInitReplicaRecovery(datanode, rBlock);
      if (info != null &&
          info.getGenerationStamp() >= block.getGenerationStamp() &&
          info.getNumBytes() > 0) {
        syncList.add(new BlockRecord(id, datanode, info));
      }
    } catch (RecoveryInProgressException ripE) {
      InterDatanodeProtocol.LOG.warn(
          "Recovery for replica " + block + " on data-node " + id
          + " is already in progress. Recovery id = "
          + rBlock.getNewGenerationStamp() + " is aborted.", ripE);
      return;
    } catch (IOException e) {
      ++errorCount;
      InterDatanodeProtocol.LOG.warn(
          "Failed to obtain replica info for block (=" + block 
          + ") from datanode (=" + id + ")", e);
    }
  }

  if (errorCount == datanodeids.length) {
    throw new IOException("All datanodes failed: block=" + block
        + ", datanodeids=" + Arrays.asList(datanodeids));
  }

  syncBlock(rBlock, syncList);
}