import org.apache.hadoop.hdfs.server.protocol.NNHAStatusHeartbeat; //导入依赖的package包/类
/**
 * The given node has reported in.  This method should:
 * 1) Record the heartbeat, so the datanode isn't timed out
 * 2) Adjust usage stats for future block allocation
 * 
 * If a substantial amount of time passed since the last datanode 
 * heartbeat then request an immediate block report.  
 * 
 * @return an array of datanode commands 
 * @throws IOException
 */
HeartbeatResponse handleHeartbeat(DatanodeRegistration nodeReg,
    StorageReport[] reports, long cacheCapacity, long cacheUsed,
    int xceiverCount, int xmitsInProgress, int failedVolumes,
    VolumeFailureSummary volumeFailureSummary) throws IOException {
  readLock();
  try {
    //get datanode commands
    final int maxTransfer = blockManager.getMaxReplicationStreams()
        - xmitsInProgress;
    DatanodeCommand[] cmds = blockManager.getDatanodeManager().handleHeartbeat(
        nodeReg, reports, blockPoolId, cacheCapacity, cacheUsed,
        xceiverCount, maxTransfer, failedVolumes, volumeFailureSummary);
    
    //create ha status
    final NNHAStatusHeartbeat haState = new NNHAStatusHeartbeat(
        haContext.getState().getServiceState(),
        getFSImage().getLastAppliedOrWrittenTxId());

    return new HeartbeatResponse(cmds, haState, rollingUpgradeInfo);
  } finally {
    readUnlock();
  }
}
 

import org.apache.hadoop.hdfs.server.protocol.NNHAStatusHeartbeat; //导入依赖的package包/类
public static NNHAStatusHeartbeatProto convert(NNHAStatusHeartbeat hb) {
  if (hb == null) return null;
  NNHAStatusHeartbeatProto.Builder builder =
    NNHAStatusHeartbeatProto.newBuilder();
  switch (hb.getState()) {
    case ACTIVE:
      builder.setState(NNHAStatusHeartbeatProto.State.ACTIVE);
      break;
    case STANDBY:
      builder.setState(NNHAStatusHeartbeatProto.State.STANDBY);
      break;
    default:
      throw new IllegalArgumentException("Unexpected NNHAStatusHeartbeat.State:" +
          hb.getState());
  }
  builder.setTxid(hb.getTxId());
  return builder.build();
}
 

import org.apache.hadoop.hdfs.server.protocol.NNHAStatusHeartbeat; //导入依赖的package包/类
/**
 * Set up a mock NN with the bare minimum for a DN to register to it.
 */
private DatanodeProtocolClientSideTranslatorPB setupNNMock(int nnIdx)
    throws Exception {
  DatanodeProtocolClientSideTranslatorPB mock =
      Mockito.mock(DatanodeProtocolClientSideTranslatorPB.class);
  Mockito.doReturn(new NamespaceInfo(1, FAKE_CLUSTERID, FAKE_BPID, 0))
      .when(mock).versionRequest();
  
  Mockito.doReturn(DFSTestUtil.getLocalDatanodeRegistration())
    .when(mock).registerDatanode(Mockito.any(DatanodeRegistration.class));
  
  Mockito.doAnswer(new HeartbeatAnswer(nnIdx))
    .when(mock).sendHeartbeat(
        Mockito.any(DatanodeRegistration.class),
        Mockito.any(StorageReport[].class),
        Mockito.anyLong(),
        Mockito.anyLong(),
        Mockito.anyInt(),
        Mockito.anyInt(),
        Mockito.anyInt(),
        Mockito.any(VolumeFailureSummary.class));
  mockHaStatuses[nnIdx] = new NNHAStatusHeartbeat(HAServiceState.STANDBY, 0);
  return mock;
}
 

import org.apache.hadoop.hdfs.server.protocol.NNHAStatusHeartbeat; //导入依赖的package包/类
public static NNHAStatusHeartbeatProto convert(NNHAStatusHeartbeat hb) {
  if (hb == null) return null;
  NNHAStatusHeartbeatProto.Builder builder =
    NNHAStatusHeartbeatProto.newBuilder();
  switch (hb.getState()) {
    case ACTIVE:
      builder.setState(HAServiceProtocolProtos.HAServiceStateProto.ACTIVE);
      break;
    case STANDBY:
      builder.setState(HAServiceProtocolProtos.HAServiceStateProto.STANDBY);
      break;
    default:
      throw new IllegalArgumentException("Unexpected NNHAStatusHeartbeat.State:" +
          hb.getState());
  }
  builder.setTxid(hb.getTxId());
  return builder.build();
}
 

import org.apache.hadoop.hdfs.server.protocol.NNHAStatusHeartbeat; //导入依赖的package包/类
/**
 * Set up a mock NN with the bare minimum for a DN to register to it.
 */
private DatanodeProtocolClientSideTranslatorPB setupNNMock(int nnIdx)
    throws Exception {
  DatanodeProtocolClientSideTranslatorPB mock =
      Mockito.mock(DatanodeProtocolClientSideTranslatorPB.class);
  Mockito.doReturn(new NamespaceInfo(1, FAKE_CLUSTERID, FAKE_BPID, 0))
      .when(mock).versionRequest();
  
  Mockito.doReturn(DFSTestUtil.getLocalDatanodeRegistration())
    .when(mock).registerDatanode(Mockito.any(DatanodeRegistration.class));
  
  Mockito.doAnswer(new HeartbeatAnswer(nnIdx))
    .when(mock).sendHeartbeat(
        Mockito.any(DatanodeRegistration.class),
        Mockito.any(StorageReport[].class),
        Mockito.anyLong(),
        Mockito.anyLong(),
        Mockito.anyInt(),
        Mockito.anyInt(),
        Mockito.anyInt(),
        Mockito.any(VolumeFailureSummary.class),
        Mockito.anyBoolean());
  mockHaStatuses[nnIdx] = new NNHAStatusHeartbeat(HAServiceState.STANDBY, 0);
  return mock;
}
 

import org.apache.hadoop.hdfs.server.protocol.NNHAStatusHeartbeat; //导入依赖的package包/类
/**
 * Set up a mock NN with the bare minimum for a DN to register to it.
 */
private DatanodeProtocolClientSideTranslatorPB setupNNMock(int nnIdx)
    throws Exception {
  DatanodeProtocolClientSideTranslatorPB mock =
      Mockito.mock(DatanodeProtocolClientSideTranslatorPB.class);
  Mockito.doReturn(new NamespaceInfo(1, FAKE_CLUSTERID, FAKE_BPID, 0))
      .when(mock).versionRequest();
  
  Mockito.doReturn(DFSTestUtil.getLocalDatanodeRegistration())
    .when(mock).registerDatanode(Mockito.any(DatanodeRegistration.class));
  
  Mockito.doAnswer(new HeartbeatAnswer(nnIdx))
    .when(mock).sendHeartbeat(
        Mockito.any(DatanodeRegistration.class),
        Mockito.any(StorageReport[].class),
        Mockito.anyInt(),
        Mockito.anyInt(),
        Mockito.anyInt());
  mockHaStatuses[nnIdx] = new NNHAStatusHeartbeat(HAServiceState.STANDBY, 0);
  return mock;
}
 

import org.apache.hadoop.hdfs.server.protocol.NNHAStatusHeartbeat; //导入依赖的package包/类
/**
 * The given node has reported in.  This method should:
 * 1) Record the heartbeat, so the datanode isn't timed out
 * 2) Adjust usage stats for future block allocation
 * 
 * If a substantial amount of time passed since the last datanode 
 * heartbeat then request an immediate block report.  
 * 
 * @return an array of datanode commands 
 * @throws IOException
 */
HeartbeatResponse handleHeartbeat(DatanodeRegistration nodeReg,
    StorageReport[] reports, long cacheCapacity, long cacheUsed,
    int xceiverCount, int xmitsInProgress, int failedVolumes)
      throws IOException {
  readLock();
  try {
    //get datanode commands
    final int maxTransfer = blockManager.getMaxReplicationStreams()
        - xmitsInProgress;
    DatanodeCommand[] cmds = blockManager.getDatanodeManager().handleHeartbeat(
        nodeReg, reports, blockPoolId, cacheCapacity, cacheUsed,
        xceiverCount, maxTransfer, failedVolumes);
    
    //create ha status
    final NNHAStatusHeartbeat haState = new NNHAStatusHeartbeat(
        haContext.getState().getServiceState(),
        getFSImage().getLastAppliedOrWrittenTxId());

    return new HeartbeatResponse(cmds, haState, rollingUpgradeInfo);
  } finally {
    readUnlock();
  }
}
 

import org.apache.hadoop.hdfs.server.protocol.NNHAStatusHeartbeat; //导入依赖的package包/类
/**
 * Set up a mock NN with the bare minimum for a DN to register to it.
 */
private DatanodeProtocolClientSideTranslatorPB setupNNMock(int nnIdx)
    throws Exception {
  DatanodeProtocolClientSideTranslatorPB mock =
      Mockito.mock(DatanodeProtocolClientSideTranslatorPB.class);
  Mockito.doReturn(new NamespaceInfo(1, FAKE_CLUSTERID, FAKE_BPID, 0))
      .when(mock).versionRequest();
  
  Mockito.doReturn(DFSTestUtil.getLocalDatanodeRegistration())
    .when(mock).registerDatanode(Mockito.any(DatanodeRegistration.class));
  
  Mockito.doAnswer(new HeartbeatAnswer(nnIdx))
    .when(mock).sendHeartbeat(
        Mockito.any(DatanodeRegistration.class),
        Mockito.any(StorageReport[].class),
        Mockito.anyLong(),
        Mockito.anyLong(),
        Mockito.anyInt(),
        Mockito.anyInt(),
        Mockito.anyInt());
  mockHaStatuses[nnIdx] = new NNHAStatusHeartbeat(HAServiceState.STANDBY, 0);
  return mock;
}
 

import org.apache.hadoop.hdfs.server.protocol.NNHAStatusHeartbeat; //导入依赖的package包/类
/**
 * This test case test the {@link BPOfferService#reportBadBlocks} method
 * such that if call to standby namenode times out then that should not 
 * affect the active namenode heartbeat processing since this function 
 * are in writeLock.
 * @throws Exception
 */
@Test
public void testReportBadBlockWhenStandbyNNTimesOut() throws Exception {
  BPOfferService bpos = setupBPOSForNNs(mockNN1, mockNN2);
  bpos.start();
  try {
    waitForInitialization(bpos);
    // Should start with neither NN as active.
    assertNull(bpos.getActiveNN());
    // Have NN1 claim active at txid 1
    mockHaStatuses[0] = new NNHAStatusHeartbeat(HAServiceState.ACTIVE, 1);
    bpos.triggerHeartbeatForTests();
    // Now mockNN1 is acting like active namenode and mockNN2 as Standby
    assertSame(mockNN1, bpos.getActiveNN());
    Mockito.doAnswer(new BPOfferServiceSynchronousCallAnswer(0))
       .when(mockNN1).reportBadBlocks(Mockito.any(LocatedBlock[].class));
    Mockito.doAnswer(new BPOfferServiceSynchronousCallAnswer(1))
       .when(mockNN2).reportBadBlocks(Mockito.any(LocatedBlock[].class));
    bpos.reportBadBlocks(FAKE_BLOCK, mockFSDataset.getVolume(FAKE_BLOCK)
        .getStorageID(), mockFSDataset.getVolume(FAKE_BLOCK)
        .getStorageType());
    bpos.reportBadBlocks(FAKE_BLOCK, mockFSDataset.getVolume(FAKE_BLOCK)
        .getStorageID(), mockFSDataset.getVolume(FAKE_BLOCK)
        .getStorageType());
    Thread.sleep(10000);
    long difference = secondCallTime - firstCallTime;
    assertTrue("Active namenode reportBadBlock processing should be "
        + "independent of standby namenode reportBadBlock processing ",
        difference < 5000);
  } finally {
    bpos.stop();
  }
}
 

import org.apache.hadoop.hdfs.server.protocol.NNHAStatusHeartbeat; //导入依赖的package包/类
/**
 * This test case test the {@link BPOfferService#trySendErrorReport} method
 * such that if call to standby namenode times out then that should not 
 * affect the active namenode heartbeat processing since this function 
 * are in writeLock.
 * @throws Exception
 */
@Test
public void testTrySendErrorReportWhenStandbyNNTimesOut() throws Exception {
  BPOfferService bpos = setupBPOSForNNs(mockNN1, mockNN2);
  bpos.start();
  try {
    waitForInitialization(bpos);
    // Should start with neither NN as active.
    assertNull(bpos.getActiveNN());
    // Have NN1 claim active at txid 1
    mockHaStatuses[0] = new NNHAStatusHeartbeat(HAServiceState.ACTIVE, 1);
    bpos.triggerHeartbeatForTests();
    // Now mockNN1 is acting like active namenode and mockNN2 as Standby
    assertSame(mockNN1, bpos.getActiveNN());
    Mockito.doAnswer(new BPOfferServiceSynchronousCallAnswer(0))
        .when(mockNN1).errorReport(Mockito.any(DatanodeRegistration.class),
        Mockito.anyInt(), Mockito.anyString());
    Mockito.doAnswer(new BPOfferServiceSynchronousCallAnswer(1))
        .when(mockNN2).errorReport(Mockito.any(DatanodeRegistration.class),
        Mockito.anyInt(), Mockito.anyString());
    String errorString = "Can't send invalid block " + FAKE_BLOCK;
    bpos.trySendErrorReport(DatanodeProtocol.INVALID_BLOCK, errorString);
    bpos.trySendErrorReport(DatanodeProtocol.INVALID_BLOCK, errorString);
    Thread.sleep(10000);
    long difference = secondCallTime - firstCallTime;
    assertTrue("Active namenode trySendErrorReport processing "
        + "should be independent of standby namenode trySendErrorReport"
        + " processing ", difference < 5000);
  } finally {
    bpos.stop();
  }
}
 

import org.apache.hadoop.hdfs.server.protocol.NNHAStatusHeartbeat; //导入依赖的package包/类
/**
 * This test case tests whether the {@BPServiceActor#processQueueMessages}
 * adds back the error report back to the queue when 
 * {BPServiceActorAction#reportTo} throws an IOException
 * @throws Exception
 */
@Test
public void testTrySendErrorReportWhenNNThrowsIOException() 
    throws Exception {
  BPOfferService bpos = setupBPOSForNNs(mockNN1, mockNN2);
  bpos.start();
  try {
    waitForInitialization(bpos);
    // Should start with neither NN as active.
    assertNull(bpos.getActiveNN());
    // Have NN1 claim active at txid 1
    mockHaStatuses[0] = new NNHAStatusHeartbeat(HAServiceState.ACTIVE, 1);
    bpos.triggerHeartbeatForTests();
    // Now mockNN1 is acting like active namenode and mockNN2 as Standby
    assertSame(mockNN1, bpos.getActiveNN());
    Mockito.doAnswer(new Answer<Void>() {
      // Throw an IOException when this function is first called which will
      // in turn add that errorReport back to the bpThreadQueue and let it 
      // process the next time. 
      @Override
      public Void answer(InvocationOnMock invocation) throws Throwable {
        if (firstCallTime == 0) {
          firstCallTime = Time.now();
          throw new IOException();
        } else {
          secondCallTime = Time.now();
          return null;
        }
      }
    }).when(mockNN1).errorReport(Mockito.any(DatanodeRegistration.class),
        Mockito.anyInt(), Mockito.anyString());
    String errorString = "Can't send invalid block " + FAKE_BLOCK;
    bpos.trySendErrorReport(DatanodeProtocol.INVALID_BLOCK, errorString);
    Thread.sleep(10000);
    assertTrue("Active namenode didn't add the report back to the queue "
        + "when errorReport threw IOException", secondCallTime != 0);
  } finally {
    bpos.stop();
  }
}
 

import org.apache.hadoop.hdfs.server.protocol.NNHAStatusHeartbeat; //导入依赖的package包/类
private static void setHeartbeatResponse(DatanodeCommand[] cmds)
    throws IOException {
  NNHAStatusHeartbeat ha = new NNHAStatusHeartbeat(HAServiceState.ACTIVE,
      fsImage.getLastAppliedOrWrittenTxId());
  HeartbeatResponse response = new HeartbeatResponse(cmds, ha, null);
  doReturn(response).when(spyNN).sendHeartbeat(
      (DatanodeRegistration) any(),
      (StorageReport[]) any(), anyLong(), anyLong(),
      anyInt(), anyInt(), anyInt(), (VolumeFailureSummary) any());
}
 

import org.apache.hadoop.hdfs.server.protocol.NNHAStatusHeartbeat; //导入依赖的package包/类
/**
 * The given node has reported in.  This method should:
 * 1) Record the heartbeat, so the datanode isn't timed out
 * 2) Adjust usage stats for future block allocation
 *
 * If a substantial amount of time passed since the last datanode
 * heartbeat then request an immediate block report.
 *
 * @return an array of datanode commands
 * @throws IOException
 */
HeartbeatResponse handleHeartbeat(DatanodeRegistration nodeReg,
    StorageReport[] reports, long cacheCapacity, long cacheUsed,
    int xceiverCount, int xmitsInProgress, int failedVolumes,
    VolumeFailureSummary volumeFailureSummary,
    boolean requestFullBlockReportLease) throws IOException {
  readLock();
  try {
    //get datanode commands
    final int maxTransfer = blockManager.getMaxReplicationStreams()
        - xmitsInProgress;
    DatanodeCommand[] cmds = blockManager.getDatanodeManager().handleHeartbeat(
        nodeReg, reports, getBlockPoolId(), cacheCapacity, cacheUsed,
        xceiverCount, maxTransfer, failedVolumes, volumeFailureSummary);
    long blockReportLeaseId = 0;
    if (requestFullBlockReportLease) {
      blockReportLeaseId =  blockManager.requestBlockReportLeaseId(nodeReg);
    }
    //create ha status
    final NNHAStatusHeartbeat haState = new NNHAStatusHeartbeat(
        haContext.getState().getServiceState(),
        getFSImage().getLastAppliedOrWrittenTxId());

    return new HeartbeatResponse(cmds, haState, rollingUpgradeInfo,
        blockReportLeaseId);
  } finally {
    readUnlock();
  }
}
 

import org.apache.hadoop.hdfs.server.protocol.NNHAStatusHeartbeat; //导入依赖的package包/类
/**
 * This test case test the {@link BPOfferService#reportBadBlocks} method
 * such that if call to standby namenode times out then that should not 
 * affect the active namenode heartbeat processing since this function 
 * are in writeLock.
 * @throws Exception
 */
@Test
public void testReportBadBlockWhenStandbyNNTimesOut() throws Exception {
  BPOfferService bpos = setupBPOSForNNs(mockNN1, mockNN2);
  bpos.start();
  try {
    waitForInitialization(bpos);
    // Should start with neither NN as active.
    assertNull(bpos.getActiveNN());
    // Have NN1 claim active at txid 1
    mockHaStatuses[0] = new NNHAStatusHeartbeat(HAServiceState.ACTIVE, 1);
    bpos.triggerHeartbeatForTests();
    // Now mockNN1 is acting like active namenode and mockNN2 as Standby
    assertSame(mockNN1, bpos.getActiveNN());
    Mockito.doAnswer(new BPOfferServiceSynchronousCallAnswer(0))
       .when(mockNN1).reportBadBlocks(Mockito.any(LocatedBlock[].class));
    Mockito.doAnswer(new BPOfferServiceSynchronousCallAnswer(1))
       .when(mockNN2).reportBadBlocks(Mockito.any(LocatedBlock[].class));
    bpos.reportBadBlocks(FAKE_BLOCK, mockFSDataset.getVolume(FAKE_BLOCK)
        .getStorageID(), mockFSDataset.getVolume(FAKE_BLOCK)
        .getStorageType());
    bpos.reportBadBlocks(FAKE_BLOCK, mockFSDataset.getVolume(FAKE_BLOCK)
        .getStorageID(), mockFSDataset.getVolume(FAKE_BLOCK)
        .getStorageType());
    Thread.sleep(10000);
    long difference = secondCallTime - firstCallTime;
    assertTrue("Active namenode reportBadBlock processing should be "
        + "independent of standby namenode reportBadBlock processing ",
        difference < 5000);
  } finally {
    bpos.stop();
    bpos.join();
  }
}
 

import org.apache.hadoop.hdfs.server.protocol.NNHAStatusHeartbeat; //导入依赖的package包/类
/**
 * This test case test the {@link BPOfferService#trySendErrorReport} method
 * such that if call to standby namenode times out then that should not 
 * affect the active namenode heartbeat processing since this function 
 * are in writeLock.
 * @throws Exception
 */
@Test
public void testTrySendErrorReportWhenStandbyNNTimesOut() throws Exception {
  BPOfferService bpos = setupBPOSForNNs(mockNN1, mockNN2);
  bpos.start();
  try {
    waitForInitialization(bpos);
    // Should start with neither NN as active.
    assertNull(bpos.getActiveNN());
    // Have NN1 claim active at txid 1
    mockHaStatuses[0] = new NNHAStatusHeartbeat(HAServiceState.ACTIVE, 1);
    bpos.triggerHeartbeatForTests();
    // Now mockNN1 is acting like active namenode and mockNN2 as Standby
    assertSame(mockNN1, bpos.getActiveNN());
    Mockito.doAnswer(new BPOfferServiceSynchronousCallAnswer(0))
        .when(mockNN1).errorReport(Mockito.any(DatanodeRegistration.class),
        Mockito.anyInt(), Mockito.anyString());
    Mockito.doAnswer(new BPOfferServiceSynchronousCallAnswer(1))
        .when(mockNN2).errorReport(Mockito.any(DatanodeRegistration.class),
        Mockito.anyInt(), Mockito.anyString());
    String errorString = "Can't send invalid block " + FAKE_BLOCK;
    bpos.trySendErrorReport(DatanodeProtocol.INVALID_BLOCK, errorString);
    bpos.trySendErrorReport(DatanodeProtocol.INVALID_BLOCK, errorString);
    Thread.sleep(10000);
    long difference = secondCallTime - firstCallTime;
    assertTrue("Active namenode trySendErrorReport processing "
        + "should be independent of standby namenode trySendErrorReport"
        + " processing ", difference < 5000);
  } finally {
    bpos.stop();
    bpos.join();
  }
}
 

import org.apache.hadoop.hdfs.server.protocol.NNHAStatusHeartbeat; //导入依赖的package包/类
private static void setHeartbeatResponse(DatanodeCommand[] cmds)
    throws IOException {
  NNHAStatusHeartbeat ha = new NNHAStatusHeartbeat(HAServiceState.ACTIVE,
      fsImage.getLastAppliedOrWrittenTxId());
  HeartbeatResponse response =
      new HeartbeatResponse(cmds, ha, null,
          ThreadLocalRandom.current().nextLong() | 1L);
  doReturn(response).when(spyNN).sendHeartbeat(
      (DatanodeRegistration) any(),
      (StorageReport[]) any(), anyLong(), anyLong(),
      anyInt(), anyInt(), anyInt(), (VolumeFailureSummary) any(),
      anyBoolean());
}