public void initializeBlockRecovery(BlockInfo blockInfo, long recoveryId,
      boolean startRecovery) {
    setBlockUCState(BlockUCState.UNDER_RECOVERY);
    blockRecoveryId = recoveryId;
    if (!startRecovery) {
      return;
    }
    if (replicas.length == 0) {
      NameNode.blockStateChangeLog.warn("BLOCK*" +
          " BlockUnderConstructionFeature.initializeBlockRecovery:" +
          " No blocks found, lease removed.");
      // sets primary node index and return.
      primaryNodeIndex = -1;
      return;
    }
    boolean allLiveReplicasTriedAsPrimary = true;
    for (ReplicaUnderConstruction replica : replicas) {
      // Check if all replicas have been tried or not.
      if (replica.isAlive()) {
        allLiveReplicasTriedAsPrimary = allLiveReplicasTriedAsPrimary
            && replica.getChosenAsPrimary();
      }
    }
    if (allLiveReplicasTriedAsPrimary) {
      // Just set all the replicas to be chosen whether they are alive or not.
      for (ReplicaUnderConstruction replica : replicas) {
        replica.setChosenAsPrimary(false);
      }
    }
    long mostRecentLastUpdate = 0;
    ReplicaUnderConstruction primary = null;
    primaryNodeIndex = -1;
    for (int i = 0; i < replicas.length; i++) {
      // Skip alive replicas which have been chosen for recovery.
      if (!(replicas[i].isAlive() && !replicas[i].getChosenAsPrimary())) {
        continue;
      }
      final ReplicaUnderConstruction ruc = replicas[i];
      final long lastUpdate = ruc.getExpectedStorageLocation()
          .getDatanodeDescriptor().getLastUpdateMonotonic();
      if (lastUpdate > mostRecentLastUpdate) {
        primaryNodeIndex = i;
        primary = ruc;
        mostRecentLastUpdate = lastUpdate;
      }
    }
    if (primary != null) {
      primary.getExpectedStorageLocation().getDatanodeDescriptor()
          .addBlockToBeRecovered(blockInfo);
      primary.setChosenAsPrimary(true);
      NameNode.blockStateChangeLog.debug(
          "BLOCK* {} recovery started, primary={}", this, primary);
    }
  }
@Test
  public void testInitializeBlockRecovery() throws Exception {
    DatanodeStorageInfo s1 = DFSTestUtil.createDatanodeStorageInfo("10.10.1.1", "s1");
    DatanodeDescriptor dd1 = s1.getDatanodeDescriptor();
    DatanodeStorageInfo s2 = DFSTestUtil.createDatanodeStorageInfo("10.10.1.2", "s2");
    DatanodeDescriptor dd2 = s2.getDatanodeDescriptor();
    DatanodeStorageInfo s3 = DFSTestUtil.createDatanodeStorageInfo("10.10.1.3", "s3");
    DatanodeDescriptor dd3 = s3.getDatanodeDescriptor();

    dd1.setAlive(true);
    dd2.setAlive(true);
    dd3.setAlive(true);
    BlockInfoContiguous blockInfo = new BlockInfoContiguous(
        new Block(0, 0, GenerationStamp.LAST_RESERVED_STAMP), (short) 3);
    blockInfo.convertToBlockUnderConstruction(BlockUCState.UNDER_CONSTRUCTION,
        new DatanodeStorageInfo[] {s1, s2, s3});

    // Recovery attempt #1.
    DFSTestUtil.resetLastUpdatesWithOffset(dd1, -3 * 1000);
    DFSTestUtil.resetLastUpdatesWithOffset(dd2, -1 * 1000);
    DFSTestUtil.resetLastUpdatesWithOffset(dd3, -2 * 1000);
    blockInfo.getUnderConstructionFeature().initializeBlockRecovery(blockInfo, 1, true);
    BlockInfo[] blockInfoRecovery = dd2.getLeaseRecoveryCommand(1);
    assertEquals(blockInfoRecovery[0], blockInfo);

    // Recovery attempt #2.
    DFSTestUtil.resetLastUpdatesWithOffset(dd1, -2 * 1000);
    DFSTestUtil.resetLastUpdatesWithOffset(dd2, -1 * 1000);
    DFSTestUtil.resetLastUpdatesWithOffset(dd3, -3 * 1000);
    blockInfo.getUnderConstructionFeature().initializeBlockRecovery(blockInfo, 2, true);
    blockInfoRecovery = dd1.getLeaseRecoveryCommand(1);
    assertEquals(blockInfoRecovery[0], blockInfo);

    // Recovery attempt #3.
    DFSTestUtil.resetLastUpdatesWithOffset(dd1, -2 * 1000);
    DFSTestUtil.resetLastUpdatesWithOffset(dd2, -1 * 1000);
    DFSTestUtil.resetLastUpdatesWithOffset(dd3, -3 * 1000);
    blockInfo.getUnderConstructionFeature().initializeBlockRecovery(blockInfo, 3, true);
    blockInfoRecovery = dd3.getLeaseRecoveryCommand(1);
    assertEquals(blockInfoRecovery[0], blockInfo);

    // Recovery attempt #4.
    // Reset everything. And again pick DN with most recent heart beat.
    DFSTestUtil.resetLastUpdatesWithOffset(dd1, -2 * 1000);
    DFSTestUtil.resetLastUpdatesWithOffset(dd2, -1 * 1000);
    DFSTestUtil.resetLastUpdatesWithOffset(dd3, 0);
    blockInfo.getUnderConstructionFeature().initializeBlockRecovery(blockInfo, 3, true);
    blockInfoRecovery = dd3.getLeaseRecoveryCommand(1);
    assertEquals(blockInfoRecovery[0], blockInfo);
  }