YARN-11871. Make client retry when no active sub-cluster available.

Author: liubin04

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/client/RMProxy.java

@@ -343,7 +343,7 @@ protected static RetryPolicy createRetryPolicy(Configuration conf,
 
       return RetryPolicies.failoverOnNetworkException(
           RetryPolicies.TRY_ONCE_THEN_FAIL, maxFailoverAttempts,
-          failoverSleepBaseMs, failoverSleepMaxMs);
+          maxFailoverAttempts, failoverSleepBaseMs, failoverSleepMaxMs);
     }
 
     if (rmConnectionRetryIntervalMS < 0) {

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/federation/policies/AbstractConfigurableFederationPolicy.java

@@ -18,8 +18,10 @@
 
 package org.apache.hadoop.yarn.server.federation.policies;
 
+import java.io.IOException;
 import java.util.Map;
 
+import org.apache.hadoop.ipc.RetriableException;
 import org.apache.hadoop.yarn.exceptions.YarnException;
 import org.apache.hadoop.yarn.server.federation.policies.dao.WeightedPolicyInfo;
 import org.apache.hadoop.yarn.server.federation.policies.exceptions.FederationPolicyInitializationException;
@@ -137,16 +139,16 @@ public void setPolicyContext(
    *
    * @return the map of ids to info for all active subclusters.
    *
-   * @throws YarnException if we can't get the list.
+   * @throws IOException if we can't get the list.
    */
   protected Map<SubClusterId, SubClusterInfo> getActiveSubclusters()
-      throws YarnException {
+      throws YarnException, IOException {
 
     Map<SubClusterId, SubClusterInfo> activeSubclusters =
         getPolicyContext().getFederationStateStoreFacade().getSubClusters(true);
 
     if (activeSubclusters == null || activeSubclusters.size() < 1) {
-      throw new NoActiveSubclustersException(
+      throw new RetriableException(
           "Zero active subclusters, cannot pick where to send job.");
     }
     return activeSubclusters;

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/federation/policies/FederationPolicyUtils.java

@@ -17,6 +17,7 @@
 
 package org.apache.hadoop.yarn.server.federation.policies;
 
+import java.io.IOException;
 import java.nio.ByteBuffer;
 import java.nio.charset.StandardCharsets;
 import java.util.ArrayList;
@@ -25,6 +26,7 @@
 
 import org.apache.hadoop.classification.InterfaceAudience.Private;
 import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.ipc.RetriableException;
 import org.apache.hadoop.yarn.conf.YarnConfiguration;
 import org.apache.hadoop.yarn.exceptions.YarnException;
 import org.apache.hadoop.yarn.server.federation.policies.amrmproxy.FederationAMRMProxyPolicy;
@@ -185,12 +187,12 @@ public static FederationAMRMProxyPolicy loadAMRMPolicy(String queue,
    * @param blackListSubClusters the list of subClusters as identified by
    *          {@link SubClusterId} to blackList from the selection of the home
    *          subCluster.
-   * @throws FederationPolicyException if there are no usable subclusters.
+   * @throws IOException if there are no usable subclusters.
    */
   public static void validateSubClusterAvailability(
       Collection<SubClusterId> activeSubClusters,
       Collection<SubClusterId> blackListSubClusters)
-      throws FederationPolicyException {
+      throws IOException {
     if (activeSubClusters != null && !activeSubClusters.isEmpty()) {
       if (blackListSubClusters == null) {
         return;
@@ -202,7 +204,7 @@ public static void validateSubClusterAvailability(
         }
       }
     }
-    throw new FederationPolicyException(
+    throw new RetriableException(
         FederationPolicyUtils.NO_ACTIVE_SUBCLUSTER_AVAILABLE);
   }
 

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/federation/policies/RouterPolicyFacade.java

@@ -17,6 +17,7 @@
 
 package org.apache.hadoop.yarn.server.federation.policies;
 
+import java.io.IOException;
 import java.nio.ByteBuffer;
 import java.nio.charset.StandardCharsets;
 import java.util.List;
@@ -127,7 +128,7 @@ public RouterPolicyFacade(Configuration conf,
    */
   public SubClusterId getHomeSubcluster(
       ApplicationSubmissionContext appSubmissionContext,
-      List<SubClusterId> blackListSubClusters) throws YarnException {
+      List<SubClusterId> blackListSubClusters) throws YarnException, IOException {
 
     // the maps are concurrent, but we need to protect from reset()
     // reinitialization mid-execution by creating a new reference local to this
@@ -233,7 +234,7 @@ public synchronized void reset() {
    *           valid sub-cluster id could be found for this reservation.
    */
   public SubClusterId getReservationHomeSubCluster(
-      ReservationSubmissionRequest request) throws YarnException {
+      ReservationSubmissionRequest request) throws YarnException, IOException {
 
     // the maps are concurrent, but we need to protect from reset()
     // reinitialization mid-execution by creating a new reference local to this

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/federation/policies/amrmproxy/BroadcastAMRMProxyPolicy.java

@@ -18,6 +18,7 @@
 
 package org.apache.hadoop.yarn.server.federation.policies.amrmproxy;
 
+import java.io.IOException;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
@@ -51,7 +52,7 @@ public void reinitialize(
   @Override
   public Map<SubClusterId, List<ResourceRequest>> splitResourceRequests(
       List<ResourceRequest> resourceRequests,
-      Set<SubClusterId> timedOutSubClusters) throws YarnException {
+      Set<SubClusterId> timedOutSubClusters) throws YarnException, IOException {
 
     Map<SubClusterId, SubClusterInfo> activeSubclusters =
         getActiveSubclusters();

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/federation/policies/amrmproxy/FederationAMRMProxyPolicy.java

@@ -17,6 +17,7 @@
 
 package org.apache.hadoop.yarn.server.federation.policies.amrmproxy;
 
+import java.io.IOException;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
@@ -49,7 +50,7 @@ public interface FederationAMRMProxyPolicy
    */
   Map<SubClusterId, List<ResourceRequest>> splitResourceRequests(
       List<ResourceRequest> resourceRequests,
-      Set<SubClusterId> timedOutSubClusters) throws YarnException;
+      Set<SubClusterId> timedOutSubClusters) throws YarnException, IOException;
 
   /**
    * This method should be invoked to notify the policy about responses being

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/federation/policies/amrmproxy/HomeAMRMProxyPolicy.java

@@ -18,12 +18,14 @@
 
 package org.apache.hadoop.yarn.server.federation.policies.amrmproxy;
 
+import java.io.IOException;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
 
+import org.apache.hadoop.ipc.RetriableException;
 import org.apache.hadoop.yarn.api.records.ResourceRequest;
 import org.apache.hadoop.yarn.exceptions.YarnException;
 import org.apache.hadoop.yarn.server.federation.policies.FederationPolicyInitializationContext;
@@ -57,14 +59,14 @@ public void reinitialize(
   @Override
   public Map<SubClusterId, List<ResourceRequest>> splitResourceRequests(
       List<ResourceRequest> resourceRequests,
-      Set<SubClusterId> timedOutSubClusters) throws YarnException {
+      Set<SubClusterId> timedOutSubClusters) throws YarnException, IOException {
     if (homeSubcluster == null) {
       throw new FederationPolicyException("No home subcluster available");
     }
 
     Map<SubClusterId, SubClusterInfo> active = getActiveSubclusters();
     if (!active.containsKey(homeSubcluster)) {
-      throw new FederationPolicyException(
+      throw new RetriableException(
           "The local subcluster " + homeSubcluster + " is not active");
     }
 

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/federation/policies/amrmproxy/LocalityMulticastAMRMProxyPolicy.java

@@ -18,6 +18,7 @@
 
 package org.apache.hadoop.yarn.server.federation.policies.amrmproxy;
 
+import java.io.IOException;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.HashMap;
@@ -34,6 +35,7 @@
 
 import org.apache.commons.collections4.MapUtils;
 import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.ipc.RetriableException;
 import org.apache.hadoop.util.StringUtils;
 import org.apache.hadoop.yarn.api.protocolrecords.AllocateResponse;
 import org.apache.hadoop.yarn.api.records.EnhancedHeadroom;
@@ -46,7 +48,6 @@
 import org.apache.hadoop.yarn.server.federation.policies.dao.WeightedPolicyInfo;
 import org.apache.hadoop.yarn.server.federation.policies.exceptions.FederationPolicyException;
 import org.apache.hadoop.yarn.server.federation.policies.exceptions.FederationPolicyInitializationException;
-import org.apache.hadoop.yarn.server.federation.policies.exceptions.NoActiveSubclustersException;
 import org.apache.hadoop.yarn.server.federation.resolver.SubClusterResolver;
 import org.apache.hadoop.yarn.server.federation.store.records.SubClusterId;
 import org.apache.hadoop.yarn.server.federation.store.records.SubClusterIdInfo;
@@ -267,7 +268,7 @@ public void notifyOfResponse(SubClusterId subClusterId,
   @Override
   public Map<SubClusterId, List<ResourceRequest>> splitResourceRequests(
       List<ResourceRequest> resourceRequests,
-      Set<SubClusterId> timedOutSubClusters) throws YarnException {
+      Set<SubClusterId> timedOutSubClusters) throws YarnException, IOException {
 
     // object used to accumulate statistics about the answer, initialize with
     // active subclusters. Create a new instance per call because this method
@@ -712,7 +713,8 @@ protected final class AllocationBookkeeper {
 
     private void reinitialize(
         Map<SubClusterId, SubClusterInfo> activeSubclusters,
-        Set<SubClusterId> timedOutSubClusters, Configuration pConf) throws YarnException {
+        Set<SubClusterId> timedOutSubClusters, Configuration pConf)
+        throws YarnException, IOException {
 
       if (MapUtils.isEmpty(activeSubclusters)) {
         throw new YarnRuntimeException("null activeSubclusters received");
@@ -752,7 +754,7 @@ private void reinitialize(
         String errorMsg = "None of the subClusters enabled in this Policy (weight > 0) are "
             + "currently active we cannot forward the ResourceRequest(s)";
         if (failOnError) {
-          throw new NoActiveSubclustersException(errorMsg);
+          throw new RetriableException(errorMsg);
         } else {
           LOG.error(errorMsg + ", continuing by enabling all active subClusters.");
           activeAndEnabledSC.addAll(activeSubclusters.keySet());

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/federation/policies/router/AbstractRouterPolicy.java

@@ -18,6 +18,7 @@
 
 package org.apache.hadoop.yarn.server.federation.policies.router;
 
+import java.io.IOException;
 import java.util.List;
 import java.util.Map;
 import java.util.Collections;
@@ -84,7 +85,8 @@ public void validate(ApplicationSubmissionContext appSubmissionContext)
    * @throws YarnException if the policy fails to choose a sub-cluster
    */
   protected abstract SubClusterId chooseSubCluster(String queue,
-      Map<SubClusterId, SubClusterInfo> preSelectSubClusters) throws YarnException;
+      Map<SubClusterId, SubClusterInfo> preSelectSubClusters)
+          throws YarnException, IOException;
 
   /**
    * Filter chosen SubCluster based on reservationId.
@@ -130,11 +132,11 @@ protected Map<SubClusterId, SubClusterInfo> prefilterSubClusters(
    * @return a hash-based chosen {@link SubClusterId} that will be the "home"
    *         for this application.
    *
-   * @throws YarnException if there are no active subclusters.
+   * @throws IOException if there are no active subclusters.
    */
   @Override
   public SubClusterId getHomeSubcluster(ApplicationSubmissionContext appContext,
-      List<SubClusterId> blackLists) throws YarnException {
+      List<SubClusterId> blackLists) throws YarnException, IOException {
 
     // null checks and default-queue behavior
     validate(appContext);
@@ -169,7 +171,7 @@ public SubClusterId getHomeSubcluster(ApplicationSubmissionContext appContext,
    */
   @Override
   public SubClusterId getReservationHomeSubcluster(ReservationSubmissionRequest request)
-      throws YarnException {
+      throws YarnException, IOException {
     if (request == null) {
       throw new FederationPolicyException("The ReservationSubmissionRequest cannot be null.");
     }

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/federation/policies/router/FederationRouterPolicy.java

@@ -17,6 +17,7 @@
 
 package org.apache.hadoop.yarn.server.federation.policies.router;
 
+import java.io.IOException;
 import java.util.List;
 
 import org.apache.hadoop.yarn.api.protocolrecords.ReservationSubmissionRequest;
@@ -49,7 +50,7 @@ public interface FederationRouterPolicy extends ConfigurableFederationPolicy {
    */
   SubClusterId getHomeSubcluster(
       ApplicationSubmissionContext appSubmissionContext,
-      List<SubClusterId> blackListSubClusters) throws YarnException;
+      List<SubClusterId> blackListSubClusters) throws YarnException, IOException;
 
   /**
    * Determines the sub-cluster where a ReservationSubmissionRequest should be
@@ -61,5 +62,5 @@ SubClusterId getHomeSubcluster(
    * @throws YarnException if the policy fails to choose a sub-cluster
    */
   SubClusterId getReservationHomeSubcluster(
-      ReservationSubmissionRequest request) throws YarnException;
+      ReservationSubmissionRequest request) throws YarnException, IOException;
 }