import org.apache.hadoop.ha.HealthMonitor.State; //导入依赖的package包/类
@Test
public void testGracefulFailoverToUnhealthy() throws Exception {
  cluster.start();

  cluster.waitForActiveLockHolder(0);

  // Mark it unhealthy, wait for it to exit election
  cluster.setHealthy(1, false);
  cluster.waitForElectorState(1, ActiveStandbyElector.State.INIT);

  // Ask for failover, it should fail, because it's unhealthy
  try {
    cluster.getService(1).getZKFCProxy(conf, 5000).gracefulFailover();
    fail("Did not fail to graceful failover to unhealthy service!");
  } catch (ServiceFailedException sfe) {
    GenericTestUtils.assertExceptionContains(
        cluster.getService(1).toString() +
        " is not currently healthy.", sfe);
  }
}
 

import org.apache.hadoop.ha.HealthMonitor.State; //导入依赖的package包/类
@Before
public void setupHM() throws InterruptedException, IOException {
  Configuration conf = new Configuration();
  conf.setInt(CommonConfigurationKeys.IPC_CLIENT_CONNECT_MAX_RETRIES_KEY, 1);
  conf.setInt(CommonConfigurationKeys.HA_HM_CHECK_INTERVAL_KEY, 50);
  conf.setInt(CommonConfigurationKeys.HA_HM_CONNECT_RETRY_INTERVAL_KEY, 50);
  conf.setInt(CommonConfigurationKeys.HA_HM_SLEEP_AFTER_DISCONNECT_KEY, 50);
  
  svc = createDummyHAService();
  hm = new HealthMonitor(conf, svc) {
    @Override
    protected HAServiceProtocol createProxy() throws IOException {
      createProxyCount.incrementAndGet();
      if (throwOOMEOnCreate) {
        throw new OutOfMemoryError("oome");
      }
      return super.createProxy();
    }
  };
  LOG.info("Starting health monitor");
  hm.start();
  
  LOG.info("Waiting for HEALTHY signal");    
  waitForState(hm, HealthMonitor.State.SERVICE_HEALTHY);
}
 

import org.apache.hadoop.ha.HealthMonitor.State; //导入依赖的package包/类
@Test(timeout=15000)
public void testGracefulFailoverToUnhealthy() throws Exception {
  try {
    cluster.start();

    cluster.waitForActiveLockHolder(0);

    // Mark it unhealthy, wait for it to exit election
    cluster.setHealthy(1, false);
    cluster.waitForElectorState(1, ActiveStandbyElector.State.INIT);
    
    // Ask for failover, it should fail, because it's unhealthy
    try {
      cluster.getService(1).getZKFCProxy(conf, 5000).gracefulFailover();
      fail("Did not fail to graceful failover to unhealthy service!");
    } catch (ServiceFailedException sfe) {
      GenericTestUtils.assertExceptionContains(
          cluster.getService(1).toString() + 
          " is not currently healthy.", sfe);
    }
  } finally {
    cluster.stop();
  }
}
 

import org.apache.hadoop.ha.HealthMonitor.State; //导入依赖的package包/类
@Before
public void setupHM() throws InterruptedException, IOException {
  Configuration conf = new Configuration();
  conf.setInt(CommonConfigurationKeys.IPC_CLIENT_CONNECT_MAX_RETRIES_KEY, 1);
  conf.setInt(CommonConfigurationKeys.HA_HM_CHECK_INTERVAL_KEY, 50);
  conf.setInt(CommonConfigurationKeys.HA_HM_CONNECT_RETRY_INTERVAL_KEY, 50);
  conf.setInt(CommonConfigurationKeys.HA_HM_SLEEP_AFTER_DISCONNECT_KEY, 50);
  
  svc = new DummyHAService(HAServiceState.ACTIVE,
      new InetSocketAddress("0.0.0.0", 0), true);
  hm = new HealthMonitor(conf, svc) {
    @Override
    protected HAServiceProtocol createProxy() throws IOException {
      createProxyCount.incrementAndGet();
      if (throwOOMEOnCreate) {
        throw new OutOfMemoryError("oome");
      }
      return super.createProxy();
    }
  };
  LOG.info("Starting health monitor");
  hm.start();
  
  LOG.info("Waiting for HEALTHY signal");    
  waitForState(hm, HealthMonitor.State.SERVICE_HEALTHY);
}
 

import org.apache.hadoop.ha.HealthMonitor.State; //导入依赖的package包/类
@Before
public void setupHM() throws InterruptedException, IOException {
  Configuration conf = new Configuration();
  conf.setInt(CommonConfigurationKeys.IPC_CLIENT_CONNECT_MAX_RETRIES_KEY, 1);
  conf.setInt(CommonConfigurationKeys.HA_HM_CHECK_INTERVAL_KEY, 50);
  conf.setInt(CommonConfigurationKeys.HA_HM_CONNECT_RETRY_INTERVAL_KEY, 50);
  conf.setInt(CommonConfigurationKeys.HA_HM_SLEEP_AFTER_DISCONNECT_KEY, 50);
  
  svc = new DummyHAService(HAServiceState.ACTIVE, null);
  hm = new HealthMonitor(conf, svc) {
    @Override
    protected HAServiceProtocol createProxy() throws IOException {
      createProxyCount.incrementAndGet();
      if (throwOOMEOnCreate) {
        throw new OutOfMemoryError("oome");
      }
      return super.createProxy();
    }
  };
  LOG.info("Starting health monitor");
  hm.start();
  
  LOG.info("Waiting for HEALTHY signal");    
  waitForState(hm, HealthMonitor.State.SERVICE_HEALTHY);
}
 

import org.apache.hadoop.ha.HealthMonitor.State; //导入依赖的package包/类
/**
 * Wait until one of the following events:
 * <ul>
 * <li>Another thread publishes the results of an attempt to become active
 * using {@link #recordActiveAttempt(ActiveAttemptRecord)}</li>
 * <li>The node enters bad health status</li>
 * <li>The specified timeout elapses</li>
 * </ul>
 * 
 * @param timeoutMillis number of millis to wait
 * @return the published record, or null if the timeout elapses or the
 * service becomes unhealthy 
 * @throws InterruptedException if the thread is interrupted.
 */
private ActiveAttemptRecord waitForActiveAttempt(int timeoutMillis)
    throws InterruptedException {
  long st = System.nanoTime();
  long waitUntil = st + TimeUnit.NANOSECONDS.convert(
      timeoutMillis, TimeUnit.MILLISECONDS);
  
  do {
    // periodically check health state, because entering an
    // unhealthy state could prevent us from ever attempting to
    // become active. We can detect this and respond to the user
    // immediately.
    synchronized (this) {
      if (lastHealthState != State.SERVICE_HEALTHY) {
        // early out if service became unhealthy
        return null;
      }
    }

    synchronized (activeAttemptRecordLock) {
      if ((lastActiveAttemptRecord != null &&
          lastActiveAttemptRecord.nanoTime >= st)) {
        return lastActiveAttemptRecord;
      }
      // Only wait 1sec so that we periodically recheck the health state
      // above.
      activeAttemptRecordLock.wait(1000);
    }
  } while (System.nanoTime() < waitUntil);
  
  // Timeout elapsed.
  LOG.warn(timeoutMillis + "ms timeout elapsed waiting for an attempt " +
      "to become active");
  return null;
}
 

import org.apache.hadoop.ha.HealthMonitor.State; //导入依赖的package包/类
/**
 * Ensure that the local node is in a healthy state, and thus
 * eligible for graceful failover.
 * @throws ServiceFailedException if the node is unhealthy
 */
private synchronized void checkEligibleForFailover()
    throws ServiceFailedException {
  // Check health
  if (this.getLastHealthState() != State.SERVICE_HEALTHY) {
    throw new ServiceFailedException(
        localTarget + " is not currently healthy. " +
        "Cannot be failover target");
  }
}
 

import org.apache.hadoop.ha.HealthMonitor.State; //导入依赖的package包/类
/**
 * Test that, if the standby node is unhealthy, it doesn't try to become
 * active
 */
@Test
public void testDontFailoverToUnhealthyNode() throws Exception {
  cluster.start();

  // Make svc1 unhealthy, and wait for its FC to notice the bad health.
  cluster.setHealthy(1, false);
  cluster.waitForHealthState(1, HealthMonitor.State.SERVICE_UNHEALTHY);

  // Expire svc0
  cluster.getElector(0).preventSessionReestablishmentForTests();
  try {
    cluster.expireActiveLockHolder(0);

    LOG.info("Expired svc0's ZK session. Waiting a second to give svc1" +
        " a chance to take the lock, if it is ever going to.");
    Thread.sleep(1000);

    // Ensure that no one holds the lock.
    cluster.waitForActiveLockHolder(null);

  } finally {
    LOG.info("Allowing svc0's elector to re-establish its connection");
    cluster.getElector(0).allowSessionReestablishmentForTests();
  }
  // svc0 should get the lock again
  cluster.waitForActiveLockHolder(0);
}
 

import org.apache.hadoop.ha.HealthMonitor.State; //导入依赖的package包/类
/**
 * Test that the ZKFC successfully quits the election when it fails to
 * become active. This allows the old node to successfully fail back.
 */
@Test
public void testBecomingActiveFails() throws Exception {
  cluster.start();
  DummyHAService svc1 = cluster.getService(1);

  LOG.info("Making svc1 fail to become active");
  cluster.setFailToBecomeActive(1, true);

  LOG.info("Faking svc0 unhealthy, should NOT successfully " +
      "failover to svc1");
  cluster.setHealthy(0, false);
  cluster.waitForHealthState(0, State.SERVICE_UNHEALTHY);
  cluster.waitForActiveLockHolder(null);


  Mockito.verify(svc1.proxy, Mockito.timeout(2000).atLeastOnce())
    .transitionToActive(Mockito.<StateChangeRequestInfo>any());

  cluster.waitForHAState(0, HAServiceState.INITIALIZING);
  cluster.waitForHAState(1, HAServiceState.STANDBY);

  LOG.info("Faking svc0 healthy again, should go back to svc0");
  cluster.setHealthy(0, true);
  cluster.waitForHAState(0, HAServiceState.ACTIVE);
  cluster.waitForHAState(1, HAServiceState.STANDBY);
  cluster.waitForActiveLockHolder(0);

  // Ensure that we can fail back to svc1  once it it is able
  // to become active (e.g the admin has restarted it)
  LOG.info("Allowing svc1 to become active, expiring svc0");
  svc1.failToBecomeActive = false;
  cluster.expireAndVerifyFailover(0, 1);
}
 

import org.apache.hadoop.ha.HealthMonitor.State; //导入依赖的package包/类
/**
 * Test that the ZKFC can gracefully cede its active status.
 */
@Test
public void testCedeActive() throws Exception {
  cluster.start();
  DummyZKFC zkfc = cluster.getZkfc(0);
  // It should be in active to start.
  assertEquals(ActiveStandbyElector.State.ACTIVE,
      zkfc.getElectorForTests().getStateForTests());

  // Ask it to cede active for 3 seconds. It should respond promptly
  // (i.e. the RPC itself should not take 3 seconds!)
  ZKFCProtocol proxy = zkfc.getLocalTarget().getZKFCProxy(conf, 5000);
  long st = Time.now();
  proxy.cedeActive(3000);
  long et = Time.now();
  assertTrue("RPC to cedeActive took " + (et - st) + " ms",
      et - st < 1000);

  // Should be in "INIT" state since it's not in the election
  // at this point.
  assertEquals(ActiveStandbyElector.State.INIT,
      zkfc.getElectorForTests().getStateForTests());

  // After the prescribed 3 seconds, should go into STANDBY state,
  // since the other node in the cluster would have taken ACTIVE.
  cluster.waitForElectorState(0, ActiveStandbyElector.State.STANDBY);
  long et2 = Time.now();
  assertTrue("Should take ~3 seconds to rejoin. Only took " + (et2 - et) +
      "ms before rejoining.",
      et2 - et > 2800);
}
 

import org.apache.hadoop.ha.HealthMonitor.State; //导入依赖的package包/类
/**
 * Test which exercises all of the inputs into ZKFC. This is particularly
 * useful for running under jcarder to check for lock order violations.
 */
@Test
public void testOneOfEverything() throws Exception {
  cluster.start();

  // Failover by session expiration
  LOG.info("====== Failing over by session expiration");
  cluster.expireAndVerifyFailover(0, 1);
  cluster.expireAndVerifyFailover(1, 0);

  // Restart ZK
  LOG.info("====== Restarting server");
  stopServer();
  waitForServerDown(hostPort, CONNECTION_TIMEOUT);
  startServer();
  waitForServerUp(hostPort, CONNECTION_TIMEOUT);

  // Failover by bad health
  cluster.setHealthy(0, false);
  cluster.waitForHAState(0, HAServiceState.INITIALIZING);
  cluster.waitForHAState(1, HAServiceState.ACTIVE);
  cluster.setHealthy(1, true);
  cluster.setHealthy(0, false);
  cluster.waitForHAState(1, HAServiceState.ACTIVE);
  cluster.waitForHAState(0, HAServiceState.INITIALIZING);
  cluster.setHealthy(0, true);

  cluster.waitForHealthState(0, State.SERVICE_HEALTHY);

  // Graceful failovers
  cluster.getZkfc(1).gracefulFailoverToYou();
  cluster.getZkfc(0).gracefulFailoverToYou();
}
 

import org.apache.hadoop.ha.HealthMonitor.State; //导入依赖的package包/类
@Test(timeout=15000)
public void testMonitor() throws Exception {
  LOG.info("Mocking bad health check, waiting for UNHEALTHY");
  svc.isHealthy = false;
  waitForState(hm, HealthMonitor.State.SERVICE_UNHEALTHY);
  
  LOG.info("Returning to healthy state, waiting for HEALTHY");
  svc.isHealthy = true;
  waitForState(hm, HealthMonitor.State.SERVICE_HEALTHY);

  LOG.info("Returning an IOException, as if node went down");
  // should expect many rapid retries
  int countBefore = createProxyCount.get();
  svc.actUnreachable = true;
  waitForState(hm, HealthMonitor.State.SERVICE_NOT_RESPONDING);

  // Should retry several times
  while (createProxyCount.get() < countBefore + 3) {
    Thread.sleep(10);
  }
  
  LOG.info("Returning to healthy state, waiting for HEALTHY");
  svc.actUnreachable = false;
  waitForState(hm, HealthMonitor.State.SERVICE_HEALTHY);
  
  hm.shutdown();
  hm.join();
  assertFalse(hm.isAlive());
}
 

import org.apache.hadoop.ha.HealthMonitor.State; //导入依赖的package包/类
/**
 * Test that the proper state is propagated when the health monitor
 * sees an uncaught exception in its thread.
 */
@Test(timeout=15000)
public void testHealthMonitorDies() throws Exception {
  LOG.info("Mocking RTE in health monitor, waiting for FAILED");
  throwOOMEOnCreate = true;
  svc.actUnreachable = true;
  waitForState(hm, HealthMonitor.State.HEALTH_MONITOR_FAILED);
  hm.shutdown();
  hm.join();
  assertFalse(hm.isAlive());
}
 

import org.apache.hadoop.ha.HealthMonitor.State; //导入依赖的package包/类
/**
 * Test that, if the callback throws an RTE, this will terminate the
 * health monitor and thus change its state to FAILED
 * @throws Exception
 */
@Test(timeout=15000)
public void testCallbackThrowsRTE() throws Exception {
  hm.addCallback(new Callback() {
    @Override
    public void enteredState(State newState) {
      throw new RuntimeException("Injected RTE");
    }
  });
  LOG.info("Mocking bad health check, waiting for UNHEALTHY");
  svc.isHealthy = false;
  waitForState(hm, HealthMonitor.State.HEALTH_MONITOR_FAILED);
}
 

import org.apache.hadoop.ha.HealthMonitor.State; //导入依赖的package包/类
private void waitForState(HealthMonitor hm, State state)
    throws InterruptedException {
  long st = Time.now();
  while (Time.now() - st < 2000) {
    if (hm.getHealthState() == state) {
      return;
    }
    Thread.sleep(50);
  }
  assertEquals(state, hm.getHealthState());
}
 

import org.apache.hadoop.ha.HealthMonitor.State; //导入依赖的package包/类
/**
 * Test that, if the standby node is unhealthy, it doesn't try to become
 * active
 */
@Test(timeout=15000)
public void testDontFailoverToUnhealthyNode() throws Exception {
  try {
    cluster.start();

    // Make svc1 unhealthy, and wait for its FC to notice the bad health.
    cluster.setHealthy(1, false);
    cluster.waitForHealthState(1, HealthMonitor.State.SERVICE_UNHEALTHY);
    
    // Expire svc0
    cluster.getElector(0).preventSessionReestablishmentForTests();
    try {
      cluster.expireActiveLockHolder(0);

      LOG.info("Expired svc0's ZK session. Waiting a second to give svc1" +
          " a chance to take the lock, if it is ever going to.");
      Thread.sleep(1000);
      
      // Ensure that no one holds the lock.
      cluster.waitForActiveLockHolder(null);
      
    } finally {
      LOG.info("Allowing svc0's elector to re-establish its connection");
      cluster.getElector(0).allowSessionReestablishmentForTests();
    }
    // svc0 should get the lock again
    cluster.waitForActiveLockHolder(0);
  } finally {
    cluster.stop();
  }
}
 

import org.apache.hadoop.ha.HealthMonitor.State; //导入依赖的package包/类
/**
 * Test that the ZKFC successfully quits the election when it fails to
 * become active. This allows the old node to successfully fail back.
 */
@Test(timeout=15000)
public void testBecomingActiveFails() throws Exception {
  try {
    cluster.start();
    DummyHAService svc1 = cluster.getService(1);
    
    LOG.info("Making svc1 fail to become active");
    cluster.setFailToBecomeActive(1, true);
    
    LOG.info("Faking svc0 unhealthy, should NOT successfully " +
        "failover to svc1");
    cluster.setHealthy(0, false);
    cluster.waitForHealthState(0, State.SERVICE_UNHEALTHY);
    cluster.waitForActiveLockHolder(null);

    
    Mockito.verify(svc1.proxy, Mockito.timeout(2000).atLeastOnce())
      .transitionToActive(Mockito.<StateChangeRequestInfo>any());

    cluster.waitForHAState(0, HAServiceState.INITIALIZING);
    cluster.waitForHAState(1, HAServiceState.STANDBY);
    
    LOG.info("Faking svc0 healthy again, should go back to svc0");
    cluster.setHealthy(0, true);
    cluster.waitForHAState(0, HAServiceState.ACTIVE);
    cluster.waitForHAState(1, HAServiceState.STANDBY);
    cluster.waitForActiveLockHolder(0);
    
    // Ensure that we can fail back to svc1  once it it is able
    // to become active (e.g the admin has restarted it)
    LOG.info("Allowing svc1 to become active, expiring svc0");
    svc1.failToBecomeActive = false;
    cluster.expireAndVerifyFailover(0, 1);
  } finally {
    cluster.stop();
  }
}
 

import org.apache.hadoop.ha.HealthMonitor.State; //导入依赖的package包/类
/**
 * Test that the ZKFC can gracefully cede its active status.
 */
@Test(timeout=15000)
public void testCedeActive() throws Exception {
  try {
    cluster.start();
    DummyZKFC zkfc = cluster.getZkfc(0);
    // It should be in active to start.
    assertEquals(ActiveStandbyElector.State.ACTIVE,
        zkfc.getElectorForTests().getStateForTests());

    // Ask it to cede active for 3 seconds. It should respond promptly
    // (i.e. the RPC itself should not take 3 seconds!)
    ZKFCProtocol proxy = zkfc.getLocalTarget().getZKFCProxy(conf, 5000);
    long st = Time.now();
    proxy.cedeActive(3000);
    long et = Time.now();
    assertTrue("RPC to cedeActive took " + (et - st) + " ms",
        et - st < 1000);
    
    // Should be in "INIT" state since it's not in the election
    // at this point.
    assertEquals(ActiveStandbyElector.State.INIT,
        zkfc.getElectorForTests().getStateForTests());

    // After the prescribed 3 seconds, should go into STANDBY state,
    // since the other node in the cluster would have taken ACTIVE.
    cluster.waitForElectorState(0, ActiveStandbyElector.State.STANDBY);
    long et2 = Time.now();
    assertTrue("Should take ~3 seconds to rejoin. Only took " + (et2 - et) +
        "ms before rejoining.",
        et2 - et > 2800);      
  } finally {
    cluster.stop();
  }
}
 

import org.apache.hadoop.ha.HealthMonitor.State; //导入依赖的package包/类
/**
 * Test which exercises all of the inputs into ZKFC. This is particularly
 * useful for running under jcarder to check for lock order violations.
 */
@Test(timeout=30000)
public void testOneOfEverything() throws Exception {
  try {
    cluster.start();
    
    // Failover by session expiration
    LOG.info("====== Failing over by session expiration");
    cluster.expireAndVerifyFailover(0, 1);
    cluster.expireAndVerifyFailover(1, 0);
    
    // Restart ZK
    LOG.info("====== Restarting server");
    stopServer();
    waitForServerDown(hostPort, CONNECTION_TIMEOUT);
    startServer();
    waitForServerUp(hostPort, CONNECTION_TIMEOUT);

    // Failover by bad health
    cluster.setHealthy(0, false);
    cluster.waitForHAState(0, HAServiceState.INITIALIZING);
    cluster.waitForHAState(1, HAServiceState.ACTIVE);
    cluster.setHealthy(1, true);
    cluster.setHealthy(0, false);
    cluster.waitForHAState(1, HAServiceState.ACTIVE);
    cluster.waitForHAState(0, HAServiceState.INITIALIZING);
    cluster.setHealthy(0, true);
    
    cluster.waitForHealthState(0, State.SERVICE_HEALTHY);
    
    // Graceful failovers
    cluster.getZkfc(1).gracefulFailoverToYou();
    cluster.getZkfc(0).gracefulFailoverToYou();
  } finally {
    cluster.stop();
  }
}