YARN-11836. Fixed AM log fetching with YARN CLI (apache#7813)

* YARN-11836. Fixed AM log fetching with YARN CLI

Change-Id: I04678a04503e01e67f1075f3fc543887cf80ac47

* YARN-11836. Fixed review comments

Change-Id: Ia7a864fa77901dd63d193e0e9f0ccaab3674a75d

* YARN-11836. Fixed typos

Change-Id: I35b0904be1c12e57349cc7a473a0d70ba38c6482

* YARN-11836. Further refactorings

Change-Id: I76250565144f8f010ece1ef9d3781259fbae4c26

Author: p-szucs

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/webapp/util/WebAppUtils.java

@@ -48,13 +48,16 @@
 import org.apache.hadoop.yarn.webapp.NotFoundException;
 import org.apache.http.NameValuePair;
 import org.apache.http.client.utils.URLEncodedUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import javax.servlet.http.HttpServletRequest;
 import javax.ws.rs.container.ContainerRequestContext;
 
 @Private
 @Evolving
 public class WebAppUtils {
+  private static final Logger LOG = LoggerFactory.getLogger(WebAppUtils.class);
   public static final String WEB_APP_TRUSTSTORE_PASSWORD_KEY =
       "ssl.server.truststore.password";
   public static final String WEB_APP_KEYSTORE_PASSWORD_KEY =
@@ -107,17 +110,34 @@ public static void setNMWebAppHostNameAndPort(Configuration conf,
    */
   public static <T, R> R execOnActiveRM(Configuration conf,
       ThrowingBiFunction<String, T, R> func, T arg) throws Exception {
-    int haIndex = 0;
-    if (HAUtil.isHAEnabled(conf)) {
-      String activeRMId = RMHAUtils.findActiveRMHAId(conf);
-      if (activeRMId != null) {
-        haIndex = new ArrayList<>(HAUtil.getRMHAIds(conf)).indexOf(activeRMId);
-      } else {
-        throw new ConnectException("No Active RM available");
+    // If HA is not enabled we are running the function on the only RM that is available.
+    if (!HAUtil.isHAEnabled(conf)) {
+      String rmAddress = getRMWebAppURLWithScheme(conf, 0);
+      return func.apply(rmAddress, arg);
+    }
+
+    // In HA mode we can find the active RM if user has admin permissions to check service states.
+    // Otherwise, activeRMId will be null.
+    List<String> rmIds = (List<String>) HAUtil.getRMHAIds(conf);
+    String activeRMId = RMHAUtils.findActiveRMHAId(conf);
+    if (activeRMId != null) {
+      int activeRMIndex = rmIds.indexOf(activeRMId);
+      String rmAddress = getRMWebAppURLWithScheme(conf, activeRMIndex);
+      return func.apply(rmAddress, arg);
+    }
+
+    // If user does not have the necessary permissions we have to iterate through the RMs
+    // to find the active one.
+    for (int i = 0; i < rmIds.size(); i++) {
+      try {
+        String rmAddress = getRMWebAppURLWithScheme(conf, i);
+        return func.apply(rmAddress, arg);
+      } catch (Exception e) {
+        // Log exception and try next RM if there are any.
+        LOG.trace("Exception while connecting to RM", e);
       }
     }
-    String rm1Address = getRMWebAppURLWithScheme(conf, haIndex);
-    return func.apply(rm1Address, arg);
+    throw new ConnectException("No active RM available to execute this command");
   }
 
   /** A BiFunction which throws on Exception. */