org.apache.hadoop.io.retry.UnreliableInterface.UnreliableException#org.apache.hadoop.ipc.StandbyException源码实例Demo

@Override
public RefreshQueuesResponse refreshQueues(RefreshQueuesRequest request)
    throws YarnException, StandbyException {
  String argName = "refreshQueues";
  final String msg = "refresh queues.";
  UserGroupInformation user = checkAcls(argName);

  checkRMStatus(user.getShortUserName(), argName, msg);

  RefreshQueuesResponse response =
      recordFactory.newRecordInstance(RefreshQueuesResponse.class);
  try {
    rmContext.getScheduler().reinitialize(getConfig(), this.rmContext);
    // refresh the reservation system
    ReservationSystem rSystem = rmContext.getReservationSystem();
    if (rSystem != null) {
      rSystem.reinitialize(getConfig(), rmContext);
    }
    RMAuditLogger.logSuccess(user.getShortUserName(), argName,
        "AdminService");
    return response;
  } catch (IOException ioe) {
    throw logAndWrapException(ioe, user.getShortUserName(), argName, msg);
  }
}
 

@Override
public RefreshNodesResponse refreshNodes(RefreshNodesRequest request)
    throws YarnException, StandbyException {
  String argName = "refreshNodes";
  final String msg = "refresh nodes.";
  UserGroupInformation user = checkAcls("refreshNodes");

  checkRMStatus(user.getShortUserName(), argName, msg);

  try {
    Configuration conf =
        getConfiguration(new Configuration(false),
            YarnConfiguration.YARN_SITE_CONFIGURATION_FILE);
    rmContext.getNodesListManager().refreshNodes(conf);
    RMAuditLogger.logSuccess(user.getShortUserName(), argName,
        "AdminService");
    return recordFactory.newRecordInstance(RefreshNodesResponse.class);
  } catch (IOException ioe) {
    throw logAndWrapException(ioe, user.getShortUserName(), argName, msg);
  }
}
 

@Override
public byte[] retrievePassword(
    DelegationTokenIdentifier identifier) throws InvalidToken {
  try {
    // this check introduces inconsistency in the authentication to a
    // HA standby NN.  non-token auths are allowed into the namespace which
    // decides whether to throw a StandbyException.  tokens are a bit
    // different in that a standby may be behind and thus not yet know
    // of all tokens issued by the active NN.  the following check does
    // not allow ANY token auth, however it should allow known tokens in
    namesystem.checkOperation(OperationCategory.READ);
  } catch (StandbyException se) {
    // FIXME: this is a hack to get around changing method signatures by
    // tunneling a non-InvalidToken exception as the cause which the
    // RPC server will unwrap before returning to the client
    InvalidToken wrappedStandby = new InvalidToken("StandbyException");
    wrappedStandby.initCause(se);
    throw wrappedStandby;
  }
  return super.retrievePassword(identifier);
}
 

@Override
public byte[] retriableRetrievePassword(DelegationTokenIdentifier identifier)
    throws InvalidToken, StandbyException, RetriableException, IOException {
  namesystem.checkOperation(OperationCategory.READ);
  try {
    return super.retrievePassword(identifier);
  } catch (InvalidToken it) {
    if (namesystem.inTransitionToActive()) {
      // if the namesystem is currently in the middle of transition to 
      // active state, let client retry since the corresponding editlog may 
      // have not been applied yet
      throw new RetriableException(it);
    } else {
      throw it;
    }
  }
}
 

/**
 * Used to ensure that at least one of the given HA NNs is currently in the
 * active state..
 * 
 * @param namenodes list of RPC proxies for each NN to check.
 * @return true if at least one NN is active, false if all are in the standby state.
 * @throws IOException in the event of error.
 */
public static boolean isAtLeastOneActive(List<ClientProtocol> namenodes)
    throws IOException {
  for (ClientProtocol namenode : namenodes) {
    try {
      namenode.getFileInfo("/");
      return true;
    } catch (RemoteException re) {
      IOException cause = re.unwrapRemoteException();
      if (cause instanceof StandbyException) {
        // This is expected to happen for a standby NN.
      } else {
        throw re;
      }
    }
  }
  return false;
}
 

@Test
public void testNeverFailOver() throws UnreliableException,
    IOException, StandbyException {
  UnreliableInterface unreliable = (UnreliableInterface)RetryProxy.create(
      UnreliableInterface.class,
      newFlipFlopProxyProvider(),
      RetryPolicies.TRY_ONCE_THEN_FAIL);

  unreliable.succeedsOnceThenFailsReturningString();
  try {
    unreliable.succeedsOnceThenFailsReturningString();
    fail("should not have succeeded twice");
  } catch (UnreliableException e) {
    assertEquals("impl1", e.getMessage());
  }
}
 

@Test
public void testFailoverOnNetworkExceptionIdempotentOperation()
    throws UnreliableException, IOException, StandbyException {
  UnreliableInterface unreliable = (UnreliableInterface)RetryProxy.create(
      UnreliableInterface.class,
      newFlipFlopProxyProvider(
          TypeOfExceptionToFailWith.IO_EXCEPTION,
          TypeOfExceptionToFailWith.UNRELIABLE_EXCEPTION),
      RetryPolicies.failoverOnNetworkException(1));
  
  assertEquals("impl1", unreliable.succeedsOnceThenFailsReturningString());
  try {
    unreliable.succeedsOnceThenFailsReturningString();
    fail("should not have succeeded twice");
  } catch (IOException e) {
    // Make sure we *don't* fail over since the first implementation threw an
    // IOException and this method is not idempotent
    assertEquals("impl1", e.getMessage());
  }
  
  assertEquals("impl1", unreliable.succeedsOnceThenFailsReturningStringIdempotent());
  // Make sure we fail over since the first implementation threw an
  // IOException and this method is idempotent.
  assertEquals("impl2", unreliable.succeedsOnceThenFailsReturningStringIdempotent());
}
 

@Test
public void testSuccedsOnceThenFailOver() throws UnreliableException,
    IOException, StandbyException {
  UnreliableInterface unreliable = (UnreliableInterface)RetryProxy.create(
      UnreliableInterface.class, newFlipFlopProxyProvider(),
      new FailOverOnceOnAnyExceptionPolicy());
  
  assertEquals("impl1", unreliable.succeedsOnceThenFailsReturningString());
  assertEquals("impl2", unreliable.succeedsOnceThenFailsReturningString());
  try {
    unreliable.succeedsOnceThenFailsReturningString();
    fail("should not have succeeded more than twice");
  } catch (UnreliableException e) {
    // expected
  }
}
 

@Test
public void testNeverFailOver() throws UnreliableException,
    IOException, StandbyException {
  UnreliableInterface unreliable = (UnreliableInterface)RetryProxy.create(
      UnreliableInterface.class,
      newFlipFlopProxyProvider(),
      RetryPolicies.TRY_ONCE_THEN_FAIL);

  unreliable.succeedsOnceThenFailsReturningString();
  try {
    unreliable.succeedsOnceThenFailsReturningString();
    fail("should not have succeeded twice");
  } catch (UnreliableException e) {
    assertEquals("impl1", e.getMessage());
  }
}
 

@Test
public void testFailoverOnNetworkExceptionIdempotentOperation()
    throws UnreliableException, IOException, StandbyException {
  UnreliableInterface unreliable = (UnreliableInterface)RetryProxy.create(
      UnreliableInterface.class,
      newFlipFlopProxyProvider(
          TypeOfExceptionToFailWith.IO_EXCEPTION,
          TypeOfExceptionToFailWith.UNRELIABLE_EXCEPTION),
      RetryPolicies.failoverOnNetworkException(1));
  
  assertEquals("impl1", unreliable.succeedsOnceThenFailsReturningString());
  try {
    unreliable.succeedsOnceThenFailsReturningString();
    fail("should not have succeeded twice");
  } catch (IOException e) {
    // Make sure we *don't* fail over since the first implementation threw an
    // IOException and this method is not idempotent
    assertEquals("impl1", e.getMessage());
  }
  
  assertEquals("impl1", unreliable.succeedsOnceThenFailsReturningStringIdempotent());
  // Make sure we fail over since the first implementation threw an
  // IOException and this method is idempotent.
  assertEquals("impl2", unreliable.succeedsOnceThenFailsReturningStringIdempotent());
}
 

@Override
public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
    AtomicLong counter = opCounter.apply(method);
    Preconditions.checkState(counter != null);
    long current = counter.getAndIncrement();
    try {
        if (current > threshold) {
            NNProxy.proxyMetrics.throttledOps.incr();
            throw new StandbyException("Too many requests (" + current + "/" + threshold + "), try later");
        }
        Object ret = method.invoke(underlying, args);
        NNProxy.proxyMetrics.successOps.incr();
        return ret;
    } catch (InvocationTargetException e) {
        NNProxy.proxyMetrics.failedOps.incr();
        throw e.getCause();
    } finally {
        counter.decrementAndGet();
    }
}
 

@Override
public RefreshNodesResponse refreshNodes(RefreshNodesRequest request)
    throws YarnException, StandbyException {
  String argName = "refreshNodes";
  final String msg = "refresh nodes.";
  UserGroupInformation user = checkAcls("refreshNodes");

  checkRMStatus(user.getShortUserName(), argName, msg);

  try {
    Configuration conf =
        getConfiguration(new Configuration(false),
            YarnConfiguration.YARN_SITE_CONFIGURATION_FILE);
    rmContext.getNodesListManager().refreshNodes(conf);
    RMAuditLogger.logSuccess(user.getShortUserName(), argName,
        "AdminService");
    return recordFactory.newRecordInstance(RefreshNodesResponse.class);
  } catch (IOException ioe) {
    throw logAndWrapException(ioe, user.getShortUserName(), argName, msg);
  }
}
 

@Override
public byte[] retrievePassword(
    DelegationTokenIdentifier identifier) throws InvalidToken {
  try {
    // this check introduces inconsistency in the authentication to a
    // HA standby NN.  non-token auths are allowed into the namespace which
    // decides whether to throw a StandbyException.  tokens are a bit
    // different in that a standby may be behind and thus not yet know
    // of all tokens issued by the active NN.  the following check does
    // not allow ANY token auth, however it should allow known tokens in
    namesystem.checkOperation(OperationCategory.READ);
  } catch (StandbyException se) {
    // FIXME: this is a hack to get around changing method signatures by
    // tunneling a non-InvalidToken exception as the cause which the
    // RPC server will unwrap before returning to the client
    InvalidToken wrappedStandby = new InvalidToken("StandbyException");
    wrappedStandby.initCause(se);
    throw wrappedStandby;
  }
  return super.retrievePassword(identifier);
}
 

@Override
public byte[] retriableRetrievePassword(DelegationTokenIdentifier identifier)
    throws InvalidToken, StandbyException, RetriableException, IOException {
  namesystem.checkOperation(OperationCategory.READ);
  try {
    return super.retrievePassword(identifier);
  } catch (InvalidToken it) {
    if (namesystem.inTransitionToActive()) {
      // if the namesystem is currently in the middle of transition to 
      // active state, let client retry since the corresponding editlog may 
      // have not been applied yet
      throw new RetriableException(it);
    } else {
      throw it;
    }
  }
}
 

@Test
public void testSuccedsOnceThenFailOver() throws UnreliableException,
    IOException, StandbyException {
  UnreliableInterface unreliable = (UnreliableInterface)RetryProxy.create(
      UnreliableInterface.class, newFlipFlopProxyProvider(),
      new FailOverOnceOnAnyExceptionPolicy());
  
  assertEquals("impl1", unreliable.succeedsOnceThenFailsReturningString());
  assertEquals("impl2", unreliable.succeedsOnceThenFailsReturningString());
  try {
    unreliable.succeedsOnceThenFailsReturningString();
    fail("should not have succeeded more than twice");
  } catch (UnreliableException e) {
    // expected
  }
}
 

/**
 * Used to ensure that at least one of the given HA NNs is currently in the
 * active state..
 * 
 * @param namenodes list of RPC proxies for each NN to check.
 * @return true if at least one NN is active, false if all are in the standby state.
 * @throws IOException in the event of error.
 */
public static boolean isAtLeastOneActive(List<ClientProtocol> namenodes)
    throws IOException {
  for (ClientProtocol namenode : namenodes) {
    try {
      namenode.getFileInfo("/");
      return true;
    } catch (RemoteException re) {
      IOException cause = re.unwrapRemoteException();
      if (cause instanceof StandbyException) {
        // This is expected to happen for a standby NN.
      } else {
        throw re;
      }
    }
  }
  return false;
}
 

private void checkRMStatus(String user, String argName, String msg)
    throws StandbyException {
  if (!isRMActive()) {
    RMAuditLogger.logFailure(user, argName, "", 
        "AdminService", "ResourceManager is not active. Can not " + msg);
    throwStandbyException();
  }
}
 

@Public
@Stable
@Idempotent
public RefreshSuperUserGroupsConfigurationResponse 
refreshSuperUserGroupsConfiguration(
    RefreshSuperUserGroupsConfigurationRequest request)
throws StandbyException, YarnException, IOException;
 

@Override
public void checkOperation(HAContext context, OperationCategory op)
    throws StandbyException {
  if (op == OperationCategory.UNCHECKED ||
      (op == OperationCategory.READ && context.allowStaleReads())) {
    return;
  }
  String msg = "Operation category " + op + " is not supported in state "
      + context.getState();
  throw new StandbyException(msg);
}
 

/**
 * DFS#isInSafeMode should check the ActiveNNs safemode in HA enabled cluster. HDFS-3507
 * 
 * @throws Exception
 */
@Test
public void testIsInSafemode() throws Exception {
  // Check for the standby nn without client failover.
  NameNode nn2 = cluster.getNameNode(1);
  assertTrue("nn2 should be in standby state", nn2.isStandbyState());

  InetSocketAddress nameNodeAddress = nn2.getNameNodeAddress();
  Configuration conf = new Configuration();
  DistributedFileSystem dfs = new DistributedFileSystem();
  try {
    dfs.initialize(
        URI.create("hdfs://" + nameNodeAddress.getHostName() + ":"
            + nameNodeAddress.getPort()), conf);
    dfs.isInSafeMode();
    fail("StandBy should throw exception for isInSafeMode");
  } catch (IOException e) {
    if (e instanceof RemoteException) {
      IOException sbExcpetion = ((RemoteException) e).unwrapRemoteException();
      assertTrue("StandBy nn should not support isInSafeMode",
          sbExcpetion instanceof StandbyException);
    } else {
      throw e;
    }
  } finally {
    if (null != dfs) {
      dfs.close();
    }
  }

  // Check with Client FailOver
  cluster.transitionToStandby(0);
  cluster.transitionToActive(1);
  cluster.getNameNodeRpc(1).setSafeMode(SafeModeAction.SAFEMODE_ENTER, false);
  DistributedFileSystem dfsWithFailOver = (DistributedFileSystem) fs;
  assertTrue("ANN should be in SafeMode", dfsWithFailOver.isInSafeMode());

  cluster.getNameNodeRpc(1).setSafeMode(SafeModeAction.SAFEMODE_LEAVE, false);
  assertFalse("ANN should be out of SafeMode", dfsWithFailOver.isInSafeMode());
}
 

@Override
public String failsIfIdentifierDoesntMatch(String identifier)
    throws UnreliableException, StandbyException, IOException {
  // Wait until all threads are trying to invoke this method
  methodLatch.countDown();
  try {
    methodLatch.await();
  } catch (InterruptedException e) {
    throw new RuntimeException(e);
  }
  return super.failsIfIdentifierDoesntMatch(identifier);
}
 

@Override
public RetryAction shouldRetry(Exception e, int retries,
    int failovers, boolean isIdempotentOrAtMostOnce) throws Exception {
  if (failovers >= maxFailovers) {
    return new RetryAction(RetryAction.RetryDecision.FAIL, 0,
        "failovers (" + failovers + ") exceeded maximum allowed ("
        + maxFailovers + ")");
  }
  if (retries - failovers > maxRetries) {
    return new RetryAction(RetryAction.RetryDecision.FAIL, 0, "retries ("
        + retries + ") exceeded maximum allowed (" + maxRetries + ")");
  }
  
  if (e instanceof ConnectException ||
      e instanceof NoRouteToHostException ||
      e instanceof UnknownHostException ||
      e instanceof StandbyException ||
      e instanceof ConnectTimeoutException ||
      isWrappedStandbyException(e)) {
    return new RetryAction(RetryAction.RetryDecision.FAILOVER_AND_RETRY,
        getFailoverOrRetrySleepTime(failovers));
  } else if (e instanceof RetriableException
      || getWrappedRetriableException(e) != null) {
    // RetriableException or RetriableException wrapped 
    return new RetryAction(RetryAction.RetryDecision.RETRY,
          getFailoverOrRetrySleepTime(retries));
  } else if (e instanceof SocketException
      || (e instanceof IOException && !(e instanceof RemoteException))) {
    if (isIdempotentOrAtMostOnce) {
      return RetryAction.FAILOVER_AND_RETRY;
    } else {
      return new RetryAction(RetryAction.RetryDecision.FAIL, 0,
          "the invoked method is not idempotent, and unable to determine "
              + "whether it was invoked");
    }
  } else {
      return fallbackPolicy.shouldRetry(e, retries, failovers,
          isIdempotentOrAtMostOnce);
  }
}
 

private static boolean isWrappedStandbyException(Exception e) {
  if (!(e instanceof RemoteException)) {
    return false;
  }
  Exception unwrapped = ((RemoteException)e).unwrapRemoteException(
      StandbyException.class);
  return unwrapped instanceof StandbyException;
}
 

@Override
public String succeedsOnceThenFailsReturningString()
    throws UnreliableException, IOException, StandbyException {
  if (succeedsOnceThenFailsCount++ < 1) {
    return identifier;
  } else {
    throwAppropriateException(exceptionToFailWith, identifier);
    return null;
  }
}
 

private static void throwAppropriateException(TypeOfExceptionToFailWith eType,
    String message) throws UnreliableException, StandbyException, IOException {
  switch (eType) {
  case STANDBY_EXCEPTION:
    throw new StandbyException(message);
  case UNRELIABLE_EXCEPTION:
    throw new UnreliableException(message);
  case IO_EXCEPTION:
    throw new IOException(message);
  case REMOTE_EXCEPTION:
    throw new RemoteException(IOException.class.getName(), message);
  default:
    throw new RuntimeException(message);
  }
}
 

@Override
public String failsIfIdentifierDoesntMatch(String identifier)
    throws UnreliableException, StandbyException, IOException {
  if (this.identifier.equals(identifier)) {
    return identifier;
  } else {
    String message = "expected '" + this.identifier + "' but received '" +
        identifier + "'";
    throwAppropriateException(exceptionToFailWith, message);
    return null;
  }
}
 

@Override
public void nonIdempotentVoidFailsIfIdentifierDoesntMatch(String identifier)
    throws UnreliableException, StandbyException, IOException {
  if (this.identifier.equals(identifier)) {
    return;
  } else {
    String message = "expected '" + this.identifier + "' but received '" +
        identifier + "'";
    throwAppropriateException(exceptionToFailWith, message);
  }
}
 

@Test
public void testSucceedsTenTimesThenFailOver() throws UnreliableException,
    IOException, StandbyException {
  UnreliableInterface unreliable = (UnreliableInterface)RetryProxy.create(
      UnreliableInterface.class,
      newFlipFlopProxyProvider(),
      new FailOverOnceOnAnyExceptionPolicy());
  
  for (int i = 0; i < 10; i++) {
    assertEquals("impl1", unreliable.succeedsTenTimesThenFailsReturningString());
  }
  assertEquals("impl2", unreliable.succeedsTenTimesThenFailsReturningString());
}
 

@Test
public void testFailoverOnStandbyException()
    throws UnreliableException, IOException, StandbyException {
  UnreliableInterface unreliable = (UnreliableInterface)RetryProxy.create(
      UnreliableInterface.class,
      newFlipFlopProxyProvider(),
      RetryPolicies.failoverOnNetworkException(1));
  
  assertEquals("impl1", unreliable.succeedsOnceThenFailsReturningString());
  try {
    unreliable.succeedsOnceThenFailsReturningString();
    fail("should not have succeeded twice");
  } catch (UnreliableException e) {
    // Make sure there was no failover on normal exception.
    assertEquals("impl1", e.getMessage());
  }
  
  unreliable = (UnreliableInterface)RetryProxy
  .create(UnreliableInterface.class,
      newFlipFlopProxyProvider(
          TypeOfExceptionToFailWith.STANDBY_EXCEPTION,
          TypeOfExceptionToFailWith.UNRELIABLE_EXCEPTION),
      RetryPolicies.failoverOnNetworkException(1));
  
  assertEquals("impl1", unreliable.succeedsOnceThenFailsReturningString());
  // Make sure we fail over since the first implementation threw a StandbyException
  assertEquals("impl2", unreliable.succeedsOnceThenFailsReturningString());
}
 

@Override
public String failsIfIdentifierDoesntMatch(String identifier)
    throws UnreliableException, StandbyException, IOException {
  // Wait until all threads are trying to invoke this method
  methodLatch.countDown();
  try {
    methodLatch.await();
  } catch (InterruptedException e) {
    throw new RuntimeException(e);
  }
  return super.failsIfIdentifierDoesntMatch(identifier);
}