YARN-11342. Fix CheckStyle.

Author: slfan1989

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/conf/YarnConfiguration.java

@@ -4117,6 +4117,14 @@ public static boolean isAclEnabled(Configuration conf) {
       ROUTER_PREFIX + "submit.retry";
   public static final int DEFAULT_ROUTER_CLIENTRM_SUBMIT_RETRY = 3;
 
+  /**
+   * GetNewApplication and SubmitApplication request retry interval time.
+   */
+  public static final String ROUTER_CLIENTRM_SUBMIT_INTERVAL_TIME =
+      ROUTER_PREFIX + "submit.interval.time";
+  public static final long DEFAULT_CLIENTRM_SUBMIT_INTERVAL_TIME =
+      TimeUnit.MILLISECONDS.toMillis(10);
+
   /**
    * The interceptor class used in FederationClientInterceptor should return
    * partial ApplicationReports.

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/resources/yarn-default.xml

@@ -5047,4 +5047,13 @@
     </description>
   </property>
 
+  <property>
+    <name>yarn.router.submit.interval.time</name>
+    <value>10</value>
+    <description>
+      The interval Time between calling different subCluster requests.
+      Default is 10ms.
+    </description>
+  </property>
+
 </configuration>

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-router/src/main/java/org/apache/hadoop/yarn/server/router/RouterAuditLogger.java

@@ -196,6 +196,24 @@ public static void logFailure(String user, String operation, String perm,
     }
   }
 
+  /**
+   * Create a readable and parsable audit log string for a failed event.
+   *
+   * @param user User who made the service request.
+   * @param operation Operation requested by the user.
+   * @param perm Target permissions.
+   * @param target The target on which the operation is being performed.
+   * @param description Some additional information as to why the operation failed.
+   * @param subClusterId SubCluster Id in which operation was performed.
+   */
+  public static void logFailure(String user, String operation, String perm,
+      String target, String description, SubClusterId subClusterId) {
+    if (LOG.isInfoEnabled()) {
+      LOG.info(createFailureLog(user, operation, perm, target, description, null,
+          subClusterId));
+    }
+  }
+
   /**
    * A helper api for creating an audit log for a failure event.
    */

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-router/src/main/java/org/apache/hadoop/yarn/server/router/RouterServerUtil.java

@@ -37,7 +37,11 @@
 
 import java.lang.reflect.InvocationTargetException;
 import java.lang.reflect.Method;
-import java.util.*;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+import java.util.Map;
+import java.util.Random;
 import java.io.IOException;
 
 /**
@@ -458,21 +462,33 @@ public static void validateContainerId(String containerId)
    *
    * @param activeSubClusters List of active subClusters.
    * @param blackList blacklist.
-   * @return Active SubClusterId
+   * @return Active SubClusterId.
+   * @throws YarnException When there is no Active SubCluster,
+   * an exception will be thrown (No active SubCluster available to submit the request.)
    */
   public static SubClusterId getRandomActiveSubCluster(
       Map<SubClusterId, SubClusterInfo> activeSubClusters, List<SubClusterId> blackList)
       throws YarnException {
+
+    // Check if activeSubClusters is empty, if it is empty, we need to throw an exception
     if (MapUtils.isEmpty(activeSubClusters)) {
       logAndThrowException(FederationPolicyUtils.NO_ACTIVE_SUBCLUSTER_AVAILABLE, null);
     }
-    List<SubClusterId> list = new ArrayList<>(activeSubClusters.keySet());
+
+    // Change activeSubClusters to List
+    List<SubClusterId> subClusterIds = new ArrayList<>(activeSubClusters.keySet());
+
+    // If the blacklist is not empty, we need to remove all the subClusters in the blacklist
     if (CollectionUtils.isNotEmpty(blackList)) {
-      list.removeAll(blackList);
+      subClusterIds.removeAll(blackList);
     }
-    if (CollectionUtils.isEmpty(list)) {
+
+    // Check there are still active subcluster after removing the blacklist
+    if (CollectionUtils.isEmpty(subClusterIds)) {
       logAndThrowException(FederationPolicyUtils.NO_ACTIVE_SUBCLUSTER_AVAILABLE, null);
     }
-    return list.get(rand.nextInt(list.size()));
+
+    // Randomly choose a SubCluster
+    return subClusterIds.get(rand.nextInt(subClusterIds.size()));
   }
 }

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-router/src/main/java/org/apache/hadoop/yarn/server/router/clientrm/FederationClientInterceptor.java

@@ -178,6 +178,7 @@ public class FederationClientInterceptor
   private ThreadPoolExecutor executorService;
   private final Clock clock = new MonotonicClock();
   private boolean returnPartialReport;
+  private long submitIntervalTime;
 
   @Override
   public void init(String userName) {
@@ -209,6 +210,10 @@ public void init(String userName) {
         YarnConfiguration.ROUTER_CLIENTRM_SUBMIT_RETRY,
         YarnConfiguration.DEFAULT_ROUTER_CLIENTRM_SUBMIT_RETRY);
 
+    submitIntervalTime = conf.getTimeDuration(
+        YarnConfiguration.ROUTER_CLIENTRM_SUBMIT_INTERVAL_TIME,
+        YarnConfiguration.DEFAULT_CLIENTRM_SUBMIT_INTERVAL_TIME, TimeUnit.MILLISECONDS);
+
     clientRMProxies = new ConcurrentHashMap<>();
     routerMetrics = RouterMetrics.getMetrics();
 
@@ -315,7 +320,7 @@ public GetNewApplicationResponse getNewApplication(
       GetNewApplicationResponse response =
           ((FederationActionRetry<GetNewApplicationResponse>) (retryCount) ->
           invokeGetNewApplication(subClustersActive, blacklist, request, retryCount)).
-          runWithRetries(actualRetryNums, 100);
+          runWithRetries(actualRetryNums, submitIntervalTime);
 
       if (response != null) {
         long stopTime = clock.getTime();
@@ -343,8 +348,8 @@ public GetNewApplicationResponse getNewApplication(
    * @param retryCount number of retries.
    * @return Get NewApplicationResponse response, If the response is empty, the request fails,
    * if the response is not empty, the request is successful.
-   * @throws YarnException
-   * @throws IOException
+   * @throws YarnException yarn exception.
+   * @throws IOException io error.
    */
   private GetNewApplicationResponse invokeGetNewApplication(
       Map<SubClusterId, SubClusterInfo> subClustersActive,
@@ -358,10 +363,12 @@ private GetNewApplicationResponse invokeGetNewApplication(
       GetNewApplicationResponse response = clientRMProxy.getNewApplication(request);
       if (response != null) {
         RouterAuditLogger.logSuccess(user.getShortUserName(), GET_NEW_APP,
-            TARGET_CLIENT_RM_SERVICE, response.getApplicationId());
+            TARGET_CLIENT_RM_SERVICE, response.getApplicationId(), subClusterId);
         return response;
       }
     } catch (Exception e) {
+      RouterAuditLogger.logFailure(user.getShortUserName(), GET_NEW_APP, UNKNOWN,
+          TARGET_CLIENT_RM_SERVICE, e.getMessage(), subClusterId);
       LOG.warn("Unable to create a new ApplicationId in SubCluster {}.", subClusterId.getId(), e);
       blackList.add(subClusterId);
       throw e;
@@ -469,7 +476,7 @@ public SubmitApplicationResponse submitApplication(
       SubmitApplicationResponse response =
           ((FederationActionRetry<SubmitApplicationResponse>) (retryCount) ->
           invokeSubmitApplication(blacklist, request, retryCount)).
-          runWithRetries(actualRetryNums, 100);
+          runWithRetries(actualRetryNums, submitIntervalTime);
 
       if (response != null) {
         long stopTime = clock.getTime();

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-router/src/test/java/org/apache/hadoop/yarn/server/router/clientrm/TestFederationClientInterceptorRetry.java

@@ -37,7 +37,6 @@
 import org.apache.hadoop.yarn.api.records.Priority;
 import org.apache.hadoop.yarn.conf.YarnConfiguration;
 import org.apache.hadoop.yarn.exceptions.YarnException;
-import org.apache.hadoop.yarn.server.federation.policies.FederationPolicyUtils;
 import org.apache.hadoop.yarn.server.federation.policies.manager.UniformBroadcastPolicyManager;
 import org.apache.hadoop.yarn.server.federation.store.impl.MemoryFederationStateStore;
 import org.apache.hadoop.yarn.server.federation.store.records.ApplicationHomeSubCluster;
@@ -53,6 +52,8 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import static org.apache.hadoop.yarn.server.federation.policies.FederationPolicyUtils.NO_ACTIVE_SUBCLUSTER_AVAILABLE;
+
 /**
  * Extends the {@code BaseRouterClientRMTest} and overrides methods in order to
  * use the {@code RouterClientRMService} pipeline test cases for testing the
@@ -169,7 +170,7 @@ public void testGetNewApplicationOneBadSC()
     setupCluster(Arrays.asList(bad2));
 
     GetNewApplicationRequest request = GetNewApplicationRequest.newInstance();
-    LambdaTestUtils.intercept(YarnException.class, FederationPolicyUtils.NO_ACTIVE_SUBCLUSTER_AVAILABLE,
+    LambdaTestUtils.intercept(YarnException.class, NO_ACTIVE_SUBCLUSTER_AVAILABLE,
         () -> interceptor.getNewApplication(request));
   }
 
@@ -184,7 +185,7 @@ public void testGetNewApplicationTwoBadSCs()
     setupCluster(Arrays.asList(bad1, bad2));
 
     GetNewApplicationRequest request = GetNewApplicationRequest.newInstance();
-    LambdaTestUtils.intercept(YarnException.class, FederationPolicyUtils.NO_ACTIVE_SUBCLUSTER_AVAILABLE,
+    LambdaTestUtils.intercept(YarnException.class, NO_ACTIVE_SUBCLUSTER_AVAILABLE,
         () -> interceptor.getNewApplication(request));
   }
 
@@ -221,7 +222,7 @@ public void testSubmitApplicationOneBadSC()
         ApplicationId.newInstance(System.currentTimeMillis(), 1);
 
     final SubmitApplicationRequest request = mockSubmitApplicationRequest(appId);
-    LambdaTestUtils.intercept(YarnException.class, FederationPolicyUtils.NO_ACTIVE_SUBCLUSTER_AVAILABLE,
+    LambdaTestUtils.intercept(YarnException.class, NO_ACTIVE_SUBCLUSTER_AVAILABLE,
         () -> interceptor.submitApplication(request));
   }
 
@@ -253,7 +254,7 @@ public void testSubmitApplicationTwoBadSCs()
         ApplicationId.newInstance(System.currentTimeMillis(), 1);
 
     final SubmitApplicationRequest request = mockSubmitApplicationRequest(appId);
-    LambdaTestUtils.intercept(YarnException.class, FederationPolicyUtils.NO_ACTIVE_SUBCLUSTER_AVAILABLE,
+    LambdaTestUtils.intercept(YarnException.class, NO_ACTIVE_SUBCLUSTER_AVAILABLE,
         () -> interceptor.submitApplication(request));
   }