code: /**
   * Licensed to the Apache Software Foundation (ASF) under one
   * or more contributor license agreements.  See the NOTICE file
   * distributed with this work for additional information
   * regarding copyright ownership.  The ASF licenses this file
   * to you under the Apache License, Version 2.0 (the
   * "License"); you may not use this file except in compliance
   * with the License.  You may obtain a copy of the License at
   *
   *     http://www.apache.org/licenses/LICENSE-2.0
   *
   * Unless required by applicable law or agreed to in writing, software
   * distributed under the License is distributed on an "AS IS" BASIS,
   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   * See the License for the specific language governing permissions and
   * limitations under the License.
   */
  package org.apache.hadoop.hdfs.server.datanode;
  
  import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_DATANODE_DATA_DIR_KEY;
  
  import org.apache.hadoop.fs.FileSystem;
  import org.apache.hadoop.fs.Path;
  import org.apache.hadoop.fs.StorageType;
  import org.apache.hadoop.hdfs.HdfsConfiguration;
  import org.apache.hadoop.hdfs.MiniDFSCluster;
  import org.apache.hadoop.hdfs.server.protocol.SlowDiskReports;
  
  import static org.apache.hadoop.test.MetricsAsserts.assertCounter;
  import static org.apache.hadoop.test.MetricsAsserts.getLongCounter;
  import static org.apache.hadoop.test.MetricsAsserts.getMetrics;
  import static org.junit.Assert.assertEquals;
  import static org.junit.Assert.assertNull;
  import static org.junit.Assert.assertNotNull;
  import static org.junit.Assert.assertSame;
  import static org.junit.Assert.assertTrue;
  
  import java.io.File;
  import java.io.IOException;
  import java.net.ConnectException;
  import java.net.InetSocketAddress;
  import java.util.ArrayList;
  import java.util.Collections;
  import java.util.List;
  import java.util.Map;
  import java.util.concurrent.atomic.AtomicBoolean;
  import java.util.concurrent.atomic.AtomicInteger;
  
  import org.apache.hadoop.metrics2.MetricsRecordBuilder;
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  import org.apache.hadoop.conf.Configuration;
  import org.apache.hadoop.ha.HAServiceProtocol.HAServiceState;
  import org.apache.hadoop.hdfs.DFSTestUtil;
  import org.apache.hadoop.hdfs.protocol.Block;
  import org.apache.hadoop.hdfs.protocol.ExtendedBlock;
  import org.apache.hadoop.hdfs.protocol.LocatedBlock;
  import org.apache.hadoop.hdfs.protocolPB.DatanodeProtocolClientSideTranslatorPB;
  import org.apache.hadoop.hdfs.server.datanode.fsdataset.FsDatasetSpi;
  import org.apache.hadoop.hdfs.server.datanode.metrics.DataNodeMetrics;
  import org.apache.hadoop.hdfs.server.datanode.SimulatedFSDataset.SimulatedStorage;
  import org.apache.hadoop.hdfs.server.protocol.BlockCommand;
  import org.apache.hadoop.hdfs.server.protocol.DatanodeCommand;
  import org.apache.hadoop.hdfs.server.protocol.DatanodeProtocol;
  import org.apache.hadoop.hdfs.server.protocol.DatanodeRegistration;
  import org.apache.hadoop.hdfs.server.protocol.DatanodeStorage;
  import org.apache.hadoop.hdfs.server.protocol.HeartbeatResponse;
  import org.apache.hadoop.hdfs.server.protocol.NNHAStatusHeartbeat;
  import org.apache.hadoop.hdfs.server.protocol.NamespaceInfo;
  import org.apache.hadoop.hdfs.server.protocol.ReceivedDeletedBlockInfo;
  import org.apache.hadoop.hdfs.server.protocol.RegisterCommand;
  import org.apache.hadoop.hdfs.server.protocol.SlowPeerReports;
  import org.apache.hadoop.hdfs.server.protocol.StorageReceivedDeletedBlocks;
  import org.apache.hadoop.hdfs.server.protocol.StorageReport;
  import org.apache.hadoop.hdfs.server.protocol.VolumeFailureSummary;
  import org.apache.hadoop.hdfs.server.protocol.StorageBlockReport;
  import org.apache.hadoop.hdfs.server.protocol.BlockReportContext;
  import org.apache.hadoop.ipc.RemoteException;
  import org.apache.hadoop.ipc.StandbyException;
  import org.apache.hadoop.ipc.protobuf.RpcHeaderProtos.RpcResponseHeaderProto.RpcErrorCodeProto;
  import org.apache.hadoop.test.GenericTestUtils;
  import org.apache.hadoop.test.PathUtils;
  import org.apache.hadoop.util.Time;
  import org.apache.log4j.Level;
  import org.junit.Before;
  import org.junit.Test;
  import org.mockito.ArgumentCaptor;
  import org.mockito.Mockito;
  import org.mockito.invocation.InvocationOnMock;
  import org.mockito.stubbing.Answer;
  
  import com.google.common.base.Supplier;
  import com.google.common.collect.Lists;
  import com.google.common.collect.Maps;
  
  public class TestBPOfferService {
  
    private static final String FAKE_BPID = "fake bpid";
    private static final String FAKE_CLUSTERID = "fake cluster";
    protected static final Logger LOG = LoggerFactory.getLogger(
        TestBPOfferService.class);
    private static final ExtendedBlock FAKE_BLOCK =
      new ExtendedBlock(FAKE_BPID, 12345L);
    private static final File TEST_BUILD_DATA = PathUtils.getTestDir(TestBPOfferService.class);
    private long firstCallTime = 0; 
    private long secondCallTime = 0;
    private long firstLeaseId = 0;
    private long secondLeaseId = 0;
    private long nextFullBlockReportLeaseId = 1L;
    private int fullBlockReportCount = 0;
    private int incrBlockReportCount = 0;
  
    static {
      GenericTestUtils.setLogLevel(DataNode.LOG, Level.ALL);
    }
  
    private DatanodeProtocolClientSideTranslatorPB mockNN1;
    private DatanodeProtocolClientSideTranslatorPB mockNN2;
    private final NNHAStatusHeartbeat[] mockHaStatuses =
        new NNHAStatusHeartbeat[3];
    private final DatanodeCommand[][] datanodeCommands =
        new DatanodeCommand[3][0];
    private final int[] heartbeatCounts = new int[3];
    private DataNode mockDn;
    private FsDatasetSpi<?> mockFSDataset;
    
    @Before
    public void setupMocks() throws Exception {
      mockNN1 = setupNNMock(0);
      mockNN2 = setupNNMock(1);
  
      // Set up a mock DN with the bare-bones configuration
      // objects, etc.
      mockDn = Mockito.mock(DataNode.class);
      Mockito.doReturn(true).when(mockDn).shouldRun();
      Configuration conf = new Configuration();
      File dnDataDir = new File(new File(TEST_BUILD_DATA, "dfs"), "data");
      conf.set(DFS_DATANODE_DATA_DIR_KEY, dnDataDir.toURI().toString());
      Mockito.doReturn(conf).when(mockDn).getConf();
      Mockito.doReturn(new DNConf(mockDn)).when(mockDn).getDnConf();
      Mockito.doReturn(DataNodeMetrics.create(conf, "fake dn"))
          .when(mockDn).getMetrics();
  
      // Set up a simulated dataset with our fake BP
      mockFSDataset = Mockito.spy(new SimulatedFSDataset(null, conf));
      mockFSDataset.addBlockPool(FAKE_BPID, conf);
  
      // Wire the dataset to the DN.
      Mockito.doReturn(mockFSDataset).when(mockDn).getFSDataset();
    }
  
    /**
     * 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());
      
      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(),
            Mockito.any(SlowPeerReports.class),
            Mockito.any(SlowDiskReports.class));
      mockHaStatuses[nnIdx] = new NNHAStatusHeartbeat(HAServiceState.STANDBY, 0);
      datanodeCommands[nnIdx] = new DatanodeCommand[0];
      return mock;
    }
    
    /**
     * Mock answer for heartbeats which returns an empty set of commands
     * and the HA status for the chosen NN from the
     * {@link TestBPOfferService#mockHaStatuses} array.
     */
    private class HeartbeatAnswer implements Answer<HeartbeatResponse> {
      private final int nnIdx;
  
      HeartbeatAnswer(int nnIdx) {
        this.nnIdx = nnIdx;
      }
  
      @Override
      public HeartbeatResponse answer(InvocationOnMock invocation)
          throws Throwable {
        heartbeatCounts[nnIdx]++;
        Boolean requestFullBlockReportLease =
            (Boolean) invocation.getArguments()[8];
        long fullBlockReportLeaseId = 0;
        if (requestFullBlockReportLease) {
          fullBlockReportLeaseId = nextFullBlockReportLeaseId++;
        }
        LOG.info("fullBlockReportLeaseId=" + fullBlockReportLeaseId);
        HeartbeatResponse heartbeatResponse = new HeartbeatResponse(
            datanodeCommands[nnIdx], mockHaStatuses[nnIdx], null,
            fullBlockReportLeaseId);
        //reset the command
        datanodeCommands[nnIdx] = new DatanodeCommand[0];
        return heartbeatResponse;
      }
    }
  
  
    private class HeartbeatRegisterAnswer implements Answer<HeartbeatResponse> {
      private final int nnIdx;
  
      HeartbeatRegisterAnswer(int nnIdx) {
        this.nnIdx = nnIdx;
      }
  
      @Override
      public HeartbeatResponse answer(InvocationOnMock invocation)
          throws Throwable {
        heartbeatCounts[nnIdx]++;
        DatanodeCommand[] cmds = new DatanodeCommand[1];
        cmds[0] = new RegisterCommand();
        return new HeartbeatResponse(cmds, mockHaStatuses[nnIdx],
            null, 0L);
      }
    }
  
    private void setBlockReportCount(int count) {
      fullBlockReportCount = count;
    }
  
    private void setIncreaseBlockReportCount(int count) {
      incrBlockReportCount += count;
    }
  
    /**
     * Test that the BPOS can register to talk to two different NNs,
     * sends block reports to both, etc.
     */
    @Test
    public void testBasicFunctionality() throws Exception {
      BPOfferService bpos = setupBPOSForNNs(mockNN1, mockNN2);
      bpos.start();
      try {
        waitForBothActors(bpos);
        
        // The DN should have register to both NNs.
        Mockito.verify(mockNN1).registerDatanode(Mockito.any());
        Mockito.verify(mockNN2).registerDatanode(Mockito.any());
  
        // Should get block reports from both NNs
        waitForBlockReport(mockNN1);
        waitForBlockReport(mockNN2);
  
        // When we receive a block, it should report it to both NNs
        bpos.notifyNamenodeReceivedBlock(FAKE_BLOCK, null, "", false);
  
        ReceivedDeletedBlockInfo[] ret = waitForBlockReceived(FAKE_BLOCK, mockNN1);
        assertEquals(1, ret.length);
        assertEquals(FAKE_BLOCK.getLocalBlock(), ret[0].getBlock());
        
        ret = waitForBlockReceived(FAKE_BLOCK, mockNN2);
        assertEquals(1, ret.length);
        assertEquals(FAKE_BLOCK.getLocalBlock(), ret[0].getBlock());
  
      } finally {
        bpos.stop();
        bpos.join();
      }
    }
  
    /**
     * HDFS-15113: Test and verify missing block when re-register.
     */
    @Test
    public void testMissBlocksWhenReregister() throws Exception {
      BPOfferService bpos = setupBPOSForNNs(mockNN1, mockNN2);
      bpos.start();
      int totalTestBlocks = 4000;
      Thread addNewBlockThread = null;
      final AtomicInteger count = new AtomicInteger(0);
  
      try {
        waitForBothActors(bpos);
        waitForInitialization(bpos);
        DataNodeFaultInjector.set(new DataNodeFaultInjector() {
          public void blockUtilSendFullBlockReport() {
            try {
              GenericTestUtils.waitFor(() -> {
                if(count.get() > 2000) {
                  return true;
                }
                return false;
              }, 100, 1000);
            } catch (Exception e) {
              e.printStackTrace();
            }
          }
        });
  
        countBlockReportItems(FAKE_BLOCK, mockNN1);
        addNewBlockThread = new Thread(() -> {
          for (int i = 0; i < totalTestBlocks; i++) {
            SimulatedFSDataset fsDataset = (SimulatedFSDataset) mockFSDataset;
            SimulatedStorage simulatedStorage = fsDataset.getStorages().get(0);
            String storageId = simulatedStorage.getStorageUuid();
            ExtendedBlock b = new ExtendedBlock(bpos.getBlockPoolId(), i, 0, i);
            try {
              fsDataset.createRbw(StorageType.DEFAULT, storageId, b, false);
              bpos.notifyNamenodeReceivingBlock(b, storageId);
              fsDataset.finalizeBlock(b, false);
              count.addAndGet(1);
              Thread.sleep(1);
            } catch (Exception e) {
              e.printStackTrace();
            }
          }
        });
        addNewBlockThread.start();
  
        // Make sure that generate blocks for DataNode and IBR not empty now.
        GenericTestUtils.waitFor(() -> {
          if(count.get() > 0) {
            return true;
          }
          return false;
        }, 100, 1000);
  
        // Trigger re-register using DataNode Command.
        datanodeCommands[0] = new DatanodeCommand[]{RegisterCommand.REGISTER};
        bpos.triggerHeartbeatForTests();
  
        try {
          GenericTestUtils.waitFor(() -> {
            if(fullBlockReportCount == totalTestBlocks ||
                incrBlockReportCount == totalTestBlocks) {
              return true;
            }
            return false;
          }, 1000, 15000);
        } catch (Exception e) {}
  
        // Verify FBR/IBR count is equal to generate number.
        assertTrue(fullBlockReportCount == totalTestBlocks ||
            incrBlockReportCount == totalTestBlocks);
      } finally {
        addNewBlockThread.join();
        bpos.stop();
        bpos.join();
  
        DataNodeFaultInjector.set(new DataNodeFaultInjector() {
          public void blockUtilSendFullBlockReport() {}
        });
      }
    }
  
    @Test
    public void testLocklessBlockPoolId() throws Exception {
      BPOfferService bpos = Mockito.spy(setupBPOSForNNs(mockNN1));
  
      // bpNSInfo is not set, should take lock to check nsInfo.
      assertNull(bpos.getBlockPoolId());
      Mockito.verify(bpos).readLock();
  
      // setting the bpNSInfo should cache the bp id, thus no locking.
      Mockito.reset(bpos);
      NamespaceInfo nsInfo = new NamespaceInfo(1, FAKE_CLUSTERID, FAKE_BPID, 0);
      assertNull(bpos.setNamespaceInfo(nsInfo));
      assertEquals(FAKE_BPID, bpos.getBlockPoolId());
      Mockito.verify(bpos, Mockito.never()).readLock();
  
      // clearing the bpNSInfo should clear the cached bp id, thus requiring
      // locking to check the bpNSInfo.
      Mockito.reset(bpos);
      assertEquals(nsInfo, bpos.setNamespaceInfo(null));
      assertNull(bpos.getBlockPoolId());
      Mockito.verify(bpos).readLock();
  
      // test setting it again.
      Mockito.reset(bpos);
      assertNull(bpos.setNamespaceInfo(nsInfo));
      assertEquals(FAKE_BPID, bpos.getBlockPoolId());
      Mockito.verify(bpos, Mockito.never()).readLock();
    }
  
    /**
     * Test that DNA_INVALIDATE commands from the standby are ignored.
     */
    @Test
    public void testIgnoreDeletionsFromNonActive() throws Exception {
      BPOfferService bpos = setupBPOSForNNs(mockNN1, mockNN2);
  
      // Ask to invalidate FAKE_BLOCK when block report hits the
      // standby
      Mockito.doReturn(new BlockCommand(DatanodeProtocol.DNA_INVALIDATE,
          FAKE_BPID, new Block[] { FAKE_BLOCK.getLocalBlock() }))
          .when(mockNN2).blockReport(
              Mockito.any(),
              Mockito.eq(FAKE_BPID),
              Mockito.any(),
              Mockito.any());
  
      bpos.start();
      try {
        waitForInitialization(bpos);
        
        // Should get block reports from both NNs
        waitForBlockReport(mockNN1);
        waitForBlockReport(mockNN2);
  
      } finally {
        bpos.stop();
        bpos.join();
      }
      
      // Should ignore the delete command from the standby
      Mockito.verify(mockFSDataset, Mockito.never())
        .invalidate(Mockito.eq(FAKE_BPID), Mockito.any());
    }
  
    /**
     * Ensure that, if the two NNs configured for a block pool
     * have different block pool IDs, they will refuse to both
     * register.
     */
    @Test
    public void testNNsFromDifferentClusters() throws Exception {
      Mockito
          .doReturn(new NamespaceInfo(1, "fake foreign cluster", FAKE_BPID, 0))
          .when(mockNN1).versionRequest();
          
      BPOfferService bpos = setupBPOSForNNs(mockNN1, mockNN2);
      bpos.start();
      try {
        waitForOneToFail(bpos);
      } finally {
        bpos.stop();
        bpos.join();
      }
    }
    
    /**
     * Test that the DataNode determines the active NameNode correctly
     * based on the HA-related information in heartbeat responses.
     * See HDFS-2627.
     */
    @Test
    public void testPickActiveNameNode() 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();
        assertSame(mockNN1, bpos.getActiveNN());
  
        // NN2 claims active at a higher txid
        mockHaStatuses[1] = new NNHAStatusHeartbeat(HAServiceState.ACTIVE, 2);
        bpos.triggerHeartbeatForTests();
        assertSame(mockNN2, bpos.getActiveNN());
        
        // Even after another heartbeat from the first NN, it should
        // think NN2 is active, since it claimed a higher txid
        bpos.triggerHeartbeatForTests();
        assertSame(mockNN2, bpos.getActiveNN());
        
        // Even if NN2 goes to standby, DN shouldn't reset to talking to NN1,
        // because NN1's txid is lower than the last active txid. Instead,
        // it should consider neither active.
        mockHaStatuses[1] = new NNHAStatusHeartbeat(HAServiceState.STANDBY, 2);
        bpos.triggerHeartbeatForTests();
        assertNull(bpos.getActiveNN());
        
        // Now if NN1 goes back to a higher txid, it should be considered active
        mockHaStatuses[0] = new NNHAStatusHeartbeat(HAServiceState.ACTIVE, 3);
        bpos.triggerHeartbeatForTests();
        assertSame(mockNN1, bpos.getActiveNN());
  
      } finally {
        bpos.stop();
        bpos.join();
      }
    }
  
    /**
     * Test datanode block pool initialization error handling.
     * Failure in initializing a block pool should not cause NPE.
     */
    @Test
    public void testBPInitErrorHandling() throws Exception {
      final DataNode mockDn = Mockito.mock(DataNode.class);
      Mockito.doReturn(true).when(mockDn).shouldRun();
      Configuration conf = new Configuration();
      File dnDataDir = new File(
        new File(TEST_BUILD_DATA, "testBPInitErrorHandling"), "data");
      conf.set(DFS_DATANODE_DATA_DIR_KEY, dnDataDir.toURI().toString());
      Mockito.doReturn(conf).when(mockDn).getConf();
      Mockito.doReturn(new DNConf(mockDn)).when(mockDn).getDnConf();
      Mockito.doReturn(DataNodeMetrics.create(conf, "fake dn")).
        when(mockDn).getMetrics();
      final AtomicInteger count = new AtomicInteger();
      Mockito.doAnswer(new Answer<Void>() {
        @Override
        public Void answer(InvocationOnMock invocation) throws Throwable {
          if (count.getAndIncrement() == 0) {
            throw new IOException("faked initBlockPool exception");
          }
          // The initBlockPool is called again. Now mock init is done.
          Mockito.doReturn(mockFSDataset).when(mockDn).getFSDataset();
          return null;
        }
      }).when(mockDn).initBlockPool(Mockito.any(BPOfferService.class));
      BPOfferService bpos = setupBPOSForNNs(mockDn, mockNN1, mockNN2);
      List<BPServiceActor> actors = bpos.getBPServiceActors();
      assertEquals(2, actors.size());
      bpos.start();
      try {
        waitForInitialization(bpos);
        // even if one of the actor initialization fails, the other one will be
        // finish block report.
        waitForBlockReport(mockNN1, mockNN2);
      } finally {
        bpos.stop();
        bpos.join();
      }
    }
  
    private void waitForOneToFail(final BPOfferService bpos)
        throws Exception {
      GenericTestUtils.waitFor(new Supplier<Boolean>() {
        @Override
        public Boolean get() {
          List<BPServiceActor> actors = bpos.getBPServiceActors();
          int failedcount = 0;
          for (BPServiceActor actor : actors) {
            if (!actor.isAlive()) {
              failedcount++;
            }
          }
          return failedcount == 1;
        }
      }, 100, 10000);
    }
  
    /**
     * Create a BPOfferService which registers with and heartbeats with the
     * specified namenode proxy objects.
     * @throws IOException 
     */
    private BPOfferService setupBPOSForNNs(
        DatanodeProtocolClientSideTranslatorPB ... nns) throws IOException {
      return setupBPOSForNNs(mockDn, nns);
    }
  
    private BPOfferService setupBPOSForNNs(DataNode mockDn,
        DatanodeProtocolClientSideTranslatorPB ... nns) throws IOException {
      // Set up some fake InetAddresses, then override the connectToNN
      // function to return the corresponding proxies.
  
      final Map<InetSocketAddress, DatanodeProtocolClientSideTranslatorPB> nnMap = Maps.newLinkedHashMap();
      List<String> nnIds = Lists.newArrayListWithCapacity(nns.length);
      for (int port = 0; port < nns.length; port++) {
        nnMap.put(new InetSocketAddress(port), nns[port]);
        Mockito.doReturn(nns[port]).when(mockDn).connectToNN(
            Mockito.eq(new InetSocketAddress(port)));
        nnIds.add("nn" + port);
      }
  
      return new BPOfferService("test_ns", nnIds,
          Lists.newArrayList(nnMap.keySet()),
          Collections.<InetSocketAddress>nCopies(nnMap.size(), null), mockDn);
    }
  
    private void waitForInitialization(final BPOfferService bpos)
        throws Exception {
      GenericTestUtils.waitFor(new Supplier<Boolean>() {
        @Override
        public Boolean get() {
          return bpos.isAlive() && bpos.isInitialized();
        }
      }, 100, 10000);
    }
  
    private void waitForBothActors(final BPOfferService bpos)
        throws Exception {
      GenericTestUtils.waitFor(new Supplier<Boolean>() {
        @Override
        public Boolean get() {
          List<BPServiceActor> actors = bpos.getBPServiceActors();
  
          return bpos.isAlive() && getRegisteredActors(actors) == 2;
        }
        private int getRegisteredActors(List<BPServiceActor> actors) {
          int regActors = 0;
          for (BPServiceActor actor : actors) {
            if (actor.getBpRegistration() != null) {
              regActors++;
            }
          }
          return regActors;
        }
      }, 100, 10000);
    }
    
    private void waitForBlockReport(final DatanodeProtocolClientSideTranslatorPB mockNN)
        throws Exception {
      GenericTestUtils.waitFor(new Supplier<Boolean>() {
        @Override
        public Boolean get() {
          try {
            Mockito.verify(mockNN).blockReport(
                Mockito.any(),
                Mockito.eq(FAKE_BPID),
                Mockito.any(),
                Mockito.any());
            return true;
          } catch (Throwable t) {
            LOG.info("waiting on block report: " + t.getMessage());
            return false;
          }
        }
      }, 500, 10000);
    }
  
    private void waitForBlockReport(
        final DatanodeProtocolClientSideTranslatorPB mockNN1,
        final DatanodeProtocolClientSideTranslatorPB mockNN2)
            throws Exception {
      GenericTestUtils.waitFor(new Supplier<Boolean>() {
        @Override
        public Boolean get() {
          return get(mockNN1) || get(mockNN2);
        }
  
        private Boolean get(DatanodeProtocolClientSideTranslatorPB mockNN) {
          try {
            Mockito.verify(mockNN).blockReport(
                    Mockito.any(),
                    Mockito.eq(FAKE_BPID),
                    Mockito.any(),
                    Mockito.any());
            return true;
          } catch (Throwable t) {
            LOG.info("waiting on block report: " + t.getMessage());
            return false;
          }
        }
      }, 500, 10000);
    }
  
    private void waitForRegistration(
        final DatanodeProtocolClientSideTranslatorPB mockNN, int times)
        throws Exception {
      GenericTestUtils.waitFor(new Supplier<Boolean>() {
        @Override
        public Boolean get() {
          try {
            // The DN should have register to both NNs.
            // first called by connectToNNAndHandshake, then called by reRegister.
            Mockito.verify(mockNN, Mockito.times(2))
                .registerDatanode(Mockito.any());
            return true;
          } catch (Throwable t) {
            LOG.info("waiting on block registerDatanode: " + t.getMessage());
            return false;
          }
        }
      }, 500, 10000);
    }
  
    private ReceivedDeletedBlockInfo[] waitForBlockReceived(
        final ExtendedBlock fakeBlock,
        final DatanodeProtocolClientSideTranslatorPB mockNN) throws Exception {
      final String fakeBlockPoolId = fakeBlock.getBlockPoolId();
      final ArgumentCaptor<StorageReceivedDeletedBlocks[]> captor =
        ArgumentCaptor.forClass(StorageReceivedDeletedBlocks[].class);
      GenericTestUtils.waitFor(new Supplier<Boolean>() {
  
        @Override
        public Boolean get() {
          try {
            Mockito.verify(mockNN).blockReceivedAndDeleted(
                Mockito.any(),
                Mockito.eq(fakeBlockPoolId),
                captor.capture());
            return true;
          } catch (Throwable t) {
            return false;
          }
        }
      }, 100, 10000);
      return captor.getValue()[0].getBlocks();
    }
  
    private void setTimeForSynchronousBPOSCalls() {
      if (firstCallTime == 0) {
        firstCallTime = Time.now();
      } else {
        secondCallTime = Time.now();
      }
    }
  
    /**
     * Record blocks counts of block report and total adding blocks count of IBR
     * which assume no deleting blocks here.
     */
    private void countBlockReportItems(final ExtendedBlock fakeBlock,
        final DatanodeProtocolClientSideTranslatorPB mockNN) throws Exception {
      final String fakeBlockPoolId = fakeBlock.getBlockPoolId();
      final ArgumentCaptor<StorageBlockReport[]> captor =
          ArgumentCaptor.forClass(StorageBlockReport[].class);
  
      // Record blocks count about the last time block report.
      Mockito.doAnswer((Answer<Object>) invocation -> {
        Object[] arguments = invocation.getArguments();
        StorageBlockReport[] list = (StorageBlockReport[])arguments[2];
        setBlockReportCount(list[0].getBlocks().getNumberOfBlocks());
        return null;
      }).when(mockNN).blockReport(
          Mockito.any(),
          Mockito.eq(fakeBlockPoolId),
          captor.capture(),
          Mockito.any()
      );
  
      // Record total adding blocks count and assume no deleting blocks here.
      Mockito.doAnswer((Answer<Object>) invocation -> {
        Object[] arguments = invocation.getArguments();
        StorageReceivedDeletedBlocks[] list =
            (StorageReceivedDeletedBlocks[])arguments[2];
        setIncreaseBlockReportCount(list[0].getBlocks().length);
        return null;
      }).when(mockNN).blockReceivedAndDeleted(
          Mockito.any(),
          Mockito.eq(fakeBlockPoolId),
          Mockito.any());
    }
    
    private class BPOfferServiceSynchronousCallAnswer implements Answer<Void> {
      private final int nnIdx;
  
      public BPOfferServiceSynchronousCallAnswer(int nnIdx) {
        this.nnIdx = nnIdx;
      }
  
      // For active namenode we will record the processTime and for standby
      // namenode we will sleep for 5 seconds (This will simulate the situation
      // where the standby namenode is down ) .
      @Override
      public Void answer(InvocationOnMock invocation) throws Throwable {
        if (nnIdx == 0) {
          setTimeForSynchronousBPOSCalls();
        } else {
          Thread.sleep(5000);
        }
        return null;
      }
     }
  
    /**
     * 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());
        Mockito.doAnswer(new BPOfferServiceSynchronousCallAnswer(1))
           .when(mockNN2).reportBadBlocks(Mockito.any());
        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();
      }
    }
  
    /**
     * 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(),
            Mockito.anyInt(), Mockito.anyString());
        Mockito.doAnswer(new BPOfferServiceSynchronousCallAnswer(1))
            .when(mockNN2).errorReport(Mockito.any(),
            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();
      }
    }
    /**
     * 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());
        // 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.
        Mockito.doThrow(new IOException("Throw IOException in the first call."))
            .doAnswer((Answer<Void>) invocation -> {
              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);
        GenericTestUtils.waitFor(() -> secondCallTime != 0, 100, 20000);
        assertTrue("Active namenode didn't add the report back to the queue "
            + "when errorReport threw IOException", secondCallTime != 0);
      } finally {
        bpos.stop();
        bpos.join();
      }
    } 
  
    /**
     * This test case doesn't add the reportBadBlock request to
     * {@link BPServiceActor#bpThreadEnqueue} when the Standby namenode throws
     * {@link StandbyException}
     * @throws Exception
     */
   @Test
    public void testReportBadBlocksWhenNNThrowsStandbyException()
        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());
        // Return nothing when active Active Namenode calls reportBadBlocks
        Mockito.doNothing().when(mockNN1).reportBadBlocks
            (Mockito.any(LocatedBlock[].class));
  
        RemoteException re = new RemoteException(StandbyException.class.
            getName(), "Operation category WRITE is not supported in state "
            + "standby", RpcErrorCodeProto.ERROR_APPLICATION);
        // Return StandbyException wrapped in RemoteException when Standby NN
        // calls reportBadBlocks
        Mockito.doThrow(re).when(mockNN2).reportBadBlocks
            (Mockito.any(LocatedBlock[].class));
  
        bpos.reportBadBlocks(FAKE_BLOCK, mockFSDataset.getVolume(FAKE_BLOCK)
            .getStorageID(), mockFSDataset.getVolume(FAKE_BLOCK)
            .getStorageType());
        // Send heartbeat so that the BpServiceActor can report bad block to
        // namenode
        bpos.triggerHeartbeatForTests();
        Mockito.verify(mockNN2, Mockito.times(1))
        .reportBadBlocks(Mockito.any(LocatedBlock[].class));
  
        // Trigger another heartbeat, this will send reportBadBlock again if it
        // is present in the queue.
        bpos.triggerHeartbeatForTests();
        Mockito.verify(mockNN2, Mockito.times(1))
            .reportBadBlocks(Mockito.any(LocatedBlock[].class));
      } finally {
        bpos.stop();
        bpos.join();
      }
    }
  
    /*
     * HDFS-9917 : Standby IBR accumulation when Standby was down.
     */
    @Test
    public void testIBRClearanceForStandbyOnReRegister() throws Exception {
      final 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());
        // Return nothing when active Active Namenode gets IBRs
        Mockito.doNothing().when(mockNN1).blockReceivedAndDeleted(
            Mockito.any(DatanodeRegistration.class), Mockito.anyString(), Mockito
                .any(StorageReceivedDeletedBlocks[].class));
  
        final IOException re = new IOException(
            "Standby NN is currently not able to process IBR");
  
        final AtomicBoolean ibrReported = new AtomicBoolean(false);
        // throw exception for standby when first IBR is receieved
        Mockito.doAnswer(new Answer<Void>() {
          @Override
          public Void answer(InvocationOnMock invocation) throws Throwable {
            ibrReported.set(true);
            throw re;
          }
        }).when(mockNN2).blockReceivedAndDeleted(
            Mockito.any(DatanodeRegistration.class), Mockito.anyString(), Mockito
                .any(StorageReceivedDeletedBlocks[].class));
  
        DatanodeStorage storage = Mockito.mock(DatanodeStorage.class);
        Mockito.doReturn(storage).when(mockFSDataset).getStorage("storage0");
        // Add IBRs
        bpos.notifyNamenodeReceivedBlock(FAKE_BLOCK, null, "storage0", false);
        // Send heartbeat so that the BpServiceActor can send IBR to
        // namenode
        bpos.triggerHeartbeatForTests();
        // Wait till first IBR is received at standbyNN. Just for confirmation.
        GenericTestUtils.waitFor(new Supplier<Boolean>() {
          @Override
          public Boolean get() {
            return ibrReported.get();
          }
        }, 100, 1000);
  
        // Send register command back to Datanode to reRegister().
        // After reRegister IBRs should be cleared.
        datanodeCommands[1] = new DatanodeCommand[] { new RegisterCommand() };
        assertEquals(
            "IBR size before reRegister should be non-0", 1, getStandbyIBRSize(
                bpos));
        bpos.triggerHeartbeatForTests();
        GenericTestUtils.waitFor(new Supplier<Boolean>() {
          @Override
          public Boolean get() {
            return getStandbyIBRSize(bpos) == 0;
          }
        }, 100, 1000);
      } finally {
        bpos.stop();
        bpos.join();
      }
    }
  
    private int getStandbyIBRSize(BPOfferService bpos) {
      List<BPServiceActor> bpServiceActors = bpos.getBPServiceActors();
      for (BPServiceActor bpServiceActor : bpServiceActors) {
        if (bpServiceActor.state == HAServiceState.STANDBY) {
          return bpServiceActor.getIbrManager().getPendingIBRSize();
        }
      }
      return -1;
    }
  
     /*
      *
      */
    @Test
    public void testNNHAStateUpdateFromVersionRequest() throws Exception {
      final BPOfferService bpos = setupBPOSForNNs(mockNN1, mockNN2);
      Mockito.doReturn(true).when(mockDn).areHeartbeatsDisabledForTests();
      BPServiceActor actor = bpos.getBPServiceActors().get(0);
      bpos.start();
      waitForInitialization(bpos);
      // Should start with neither NN as active.
      assertNull(bpos.getActiveNN());
  
      // getNamespaceInfo() will not include HAServiceState
      NamespaceInfo nsInfo = mockNN1.versionRequest();
      bpos.verifyAndSetNamespaceInfo(actor, nsInfo);
  
      assertNull(bpos.getActiveNN());
  
      // Change mock so getNamespaceInfo() will include HAServiceState
      Mockito.doReturn(new NamespaceInfo(1, FAKE_CLUSTERID, FAKE_BPID, 0,
          HAServiceState.ACTIVE)).when(mockNN1).versionRequest();
  
      // Update the bpos NamespaceInfo
      nsInfo = mockNN1.versionRequest();
      bpos.verifyAndSetNamespaceInfo(actor, nsInfo);
  
      assertNotNull(bpos.getActiveNN());
  
    }
  
    @Test(timeout = 30000)
    public void testRefreshNameNodes() throws Exception {
  
      BPOfferService bpos = setupBPOSForNNs(mockDn, mockNN1, mockNN2);
  
      bpos.start();
      try {
        waitForBothActors(bpos);
  
        // The DN should have register to both NNs.
        Mockito.verify(mockNN1)
            .registerDatanode(Mockito.any());
        Mockito.verify(mockNN2)
            .registerDatanode(Mockito.any());
  
        // Should get block reports from both NNs
        waitForBlockReport(mockNN1);
        waitForBlockReport(mockNN2);
  
        // When we receive a block, it should report it to both NNs
        bpos.notifyNamenodeReceivedBlock(FAKE_BLOCK, null, "", false);
  
        ReceivedDeletedBlockInfo[] ret = waitForBlockReceived(FAKE_BLOCK,
            mockNN1);
        assertEquals(1, ret.length);
        assertEquals(FAKE_BLOCK.getLocalBlock(), ret[0].getBlock());
  
        ret = waitForBlockReceived(FAKE_BLOCK, mockNN2);
        assertEquals(1, ret.length);
        assertEquals(FAKE_BLOCK.getLocalBlock(), ret[0].getBlock());
  
        // add new standby
        DatanodeProtocolClientSideTranslatorPB mockNN3 = setupNNMock(2);
        Mockito.doReturn(mockNN3).when(mockDn)
            .connectToNN(Mockito.eq(new InetSocketAddress(2)));
  
        ArrayList<InetSocketAddress> addrs = new ArrayList<>();
        ArrayList<InetSocketAddress> lifelineAddrs = new ArrayList<>(
            addrs.size());
        // mockNN1
        addrs.add(new InetSocketAddress(0));
        lifelineAddrs.add(null);
        // mockNN3
        addrs.add(new InetSocketAddress(2));
        lifelineAddrs.add(null);
  
        ArrayList<String> nnIds = new ArrayList<>(addrs.size());
        for (int i = 0; i < addrs.size(); i++) {
          nnIds.add("nn" + i);
        }
  
        bpos.refreshNNList("serviceId", nnIds, addrs, lifelineAddrs);
  
        assertEquals(2, bpos.getBPServiceActors().size());
        // wait for handshake to run
        Thread.sleep(1000);
  
        // verify new NN registered
        Mockito.verify(mockNN3).registerDatanode(Mockito.any());
  
        // When we receive a block, it should report it to both NNs
        bpos.notifyNamenodeReceivedBlock(FAKE_BLOCK, null, "", false);
  
        // veridfy new NN recieved block report
        ret = waitForBlockReceived(FAKE_BLOCK, mockNN3);
        assertEquals(1, ret.length);
        assertEquals(FAKE_BLOCK.getLocalBlock(), ret[0].getBlock());
  
      } finally {
        bpos.stop();
        bpos.join();
      }
    }
  
    @Test(timeout = 15000)
    public void testRefreshLeaseId() throws Exception {
      Mockito.when(mockNN1.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(),
          Mockito.any(SlowPeerReports.class),
          Mockito.any(SlowDiskReports.class)))
          //heartbeat to old NN instance
          .thenAnswer(new HeartbeatAnswer(0))
          //heartbeat to new NN instance with Register Command
          .thenAnswer(new HeartbeatRegisterAnswer(0))
          .thenAnswer(new HeartbeatAnswer(0));
  
      Mockito.when(mockNN1.blockReport(
          Mockito.any(DatanodeRegistration.class),
          Mockito.anyString(),
          Mockito.any(StorageBlockReport[].class),
          Mockito.any(BlockReportContext.class)))
          .thenAnswer(
            new Answer() {
              @Override
                public Object answer(InvocationOnMock invocation)
                    throws Throwable {
                  BlockReportContext context =
                      (BlockReportContext) invocation.getArguments()[3];
                  long leaseId = context.getLeaseId();
                  LOG.info("leaseId = "+leaseId);
  
                  // leaseId == 1 means DN make block report with old leaseId
                  // just reject and wait until DN request for a new leaseId
                  if(leaseId == 1) {
                    firstLeaseId = leaseId;
                    throw new ConnectException(
                            "network is not reachable for test. ");
                  } else {
                    secondLeaseId = leaseId;
                    return null;
                  }
                }
            });
  
      BPOfferService bpos = setupBPOSForNNs(mockNN1);
      bpos.start();
  
      try {
        waitForInitialization(bpos);
        // Should call registration 2 times
        waitForRegistration(mockNN1, 2);
        assertEquals(1L, firstLeaseId);
        while(secondLeaseId != 2L) {
          Thread.sleep(1000);
        }
      } finally {
        bpos.stop();
      }
    }
  
    @Test(timeout = 15000)
    public void testCommandProcessingThread() throws Exception {
      Configuration conf = new HdfsConfiguration();
      MiniDFSCluster cluster = new MiniDFSCluster.Builder(conf).build();
      try {
        List<DataNode> datanodes = cluster.getDataNodes();
        assertEquals(datanodes.size(), 1);
        DataNode datanode = datanodes.get(0);
  
        // Try to write file and trigger NN send back command to DataNode.
        FileSystem fs = cluster.getFileSystem();
        Path file = new Path("/test");
        DFSTestUtil.createFile(fs, file, 10240L, (short)1, 0L);
  
        MetricsRecordBuilder mrb = getMetrics(datanode.getMetrics().name());
        assertTrue("Process command nums is not expected.",
            getLongCounter("NumProcessedCommands", mrb) > 0);
        assertEquals(0, getLongCounter("SumOfActorCommandQueueLength", mrb));
        // Check new metric result about processedCommandsOp.
        // One command send back to DataNode here is #FinalizeCommand.
        assertCounter("ProcessedCommandsOpNumOps", 1L, mrb);
      } finally {
        if (cluster != null) {
          cluster.shutdown();
        }
      }
    }
  }