YARN-10767. Yarn Logs Command retrying on Standby RM for 30 times. Contributed by D M Murali Krishna Reddy.
(cherry picked from commit 9a6a11c452
)
This commit is contained in:
parent
b8fc956a4d
commit
2636b630b4
|
@ -23,6 +23,7 @@ import java.util.Collection;
|
||||||
import java.util.List;
|
import java.util.List;
|
||||||
import org.apache.hadoop.classification.InterfaceAudience.Private;
|
import org.apache.hadoop.classification.InterfaceAudience.Private;
|
||||||
import org.apache.hadoop.classification.InterfaceStability.Unstable;
|
import org.apache.hadoop.classification.InterfaceStability.Unstable;
|
||||||
|
import org.apache.hadoop.conf.Configuration;
|
||||||
import org.apache.hadoop.fs.CommonConfigurationKeys;
|
import org.apache.hadoop.fs.CommonConfigurationKeys;
|
||||||
import org.apache.hadoop.ha.HAServiceProtocol;
|
import org.apache.hadoop.ha.HAServiceProtocol;
|
||||||
import org.apache.hadoop.ha.HAServiceProtocol.HAServiceState;
|
import org.apache.hadoop.ha.HAServiceProtocol.HAServiceState;
|
||||||
|
@ -35,7 +36,7 @@ import org.apache.hadoop.yarn.conf.YarnConfiguration;
|
||||||
@Unstable
|
@Unstable
|
||||||
public class RMHAUtils {
|
public class RMHAUtils {
|
||||||
|
|
||||||
public static String findActiveRMHAId(YarnConfiguration conf) {
|
public static String findActiveRMHAId(Configuration conf) {
|
||||||
YarnConfiguration yarnConf = new YarnConfiguration(conf);
|
YarnConfiguration yarnConf = new YarnConfiguration(conf);
|
||||||
Collection<String> rmIds =
|
Collection<String> rmIds =
|
||||||
yarnConf.getStringCollection(YarnConfiguration.RM_HA_IDS);
|
yarnConf.getStringCollection(YarnConfiguration.RM_HA_IDS);
|
||||||
|
|
|
@ -20,6 +20,7 @@ package org.apache.hadoop.yarn.webapp.util;
|
||||||
import static org.apache.hadoop.yarn.util.StringHelper.PATH_JOINER;
|
import static org.apache.hadoop.yarn.util.StringHelper.PATH_JOINER;
|
||||||
|
|
||||||
import java.io.IOException;
|
import java.io.IOException;
|
||||||
|
import java.net.ConnectException;
|
||||||
import java.net.InetAddress;
|
import java.net.InetAddress;
|
||||||
import java.net.InetSocketAddress;
|
import java.net.InetSocketAddress;
|
||||||
import java.net.UnknownHostException;
|
import java.net.UnknownHostException;
|
||||||
|
@ -97,24 +98,17 @@ public class WebAppUtils {
|
||||||
*/
|
*/
|
||||||
public static <T, R> R execOnActiveRM(Configuration conf,
|
public static <T, R> R execOnActiveRM(Configuration conf,
|
||||||
ThrowingBiFunction<String, T, R> func, T arg) throws Exception {
|
ThrowingBiFunction<String, T, R> func, T arg) throws Exception {
|
||||||
String rm1Address = getRMWebAppURLWithScheme(conf, 0);
|
int haIndex = 0;
|
||||||
try {
|
if (HAUtil.isHAEnabled(conf)) {
|
||||||
return func.apply(rm1Address, arg);
|
String activeRMId = RMHAUtils.findActiveRMHAId(conf);
|
||||||
} catch (Exception e) {
|
if (activeRMId != null) {
|
||||||
if (HAUtil.isHAEnabled(conf)) {
|
haIndex = new ArrayList<>(HAUtil.getRMHAIds(conf)).indexOf(activeRMId);
|
||||||
int rms = HAUtil.getRMHAIds(conf).size();
|
} else {
|
||||||
for (int i=1; i<rms; i++) {
|
throw new ConnectException("No Active RM available");
|
||||||
try {
|
|
||||||
rm1Address = getRMWebAppURLWithScheme(conf, i);
|
|
||||||
return func.apply(rm1Address, arg);
|
|
||||||
} catch (Exception e1) {
|
|
||||||
// ignore and try next one when RM is down
|
|
||||||
e = e1;
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
throw e;
|
|
||||||
}
|
}
|
||||||
|
String rm1Address = getRMWebAppURLWithScheme(conf, haIndex);
|
||||||
|
return func.apply(rm1Address, arg);
|
||||||
}
|
}
|
||||||
|
|
||||||
/** A BiFunction which throws on Exception. */
|
/** A BiFunction which throws on Exception. */
|
||||||
|
|
Loading…
Reference in New Issue