YARN-6247. Share a single instance of SubClusterResolver instead of instantiating one per AM. (Botong Huang via Subru)
(cherry picked from commit 51aeb2ce0c599176aca9466a939c3ad55df30036)
This commit is contained in:
parent
8623644f45
commit
86b2bec56e
@ -2594,6 +2594,12 @@ public static boolean isAclEnabled(Configuration conf) {
|
||||
public static final String FEDERATION_MACHINE_LIST =
|
||||
FEDERATION_PREFIX + "machine-list";
|
||||
|
||||
public static final String FEDERATION_CLUSTER_RESOLVER_CLASS =
|
||||
FEDERATION_PREFIX + "subcluster-resolver.class";
|
||||
public static final String DEFAULT_FEDERATION_CLUSTER_RESOLVER_CLASS =
|
||||
"org.apache.hadoop.yarn.server.federation.resolver."
|
||||
+ "DefaultSubClusterResolverImpl";
|
||||
|
||||
public static final String DEFAULT_FEDERATION_POLICY_KEY = "*";
|
||||
|
||||
public static final String FEDERATION_POLICY_MANAGER = FEDERATION_PREFIX
|
||||
|
@ -2701,6 +2701,13 @@
|
||||
</description>
|
||||
<name>yarn.federation.machine-list</name>
|
||||
</property>
|
||||
<property>
|
||||
<description>
|
||||
Class name for SubClusterResolver
|
||||
</description>
|
||||
<name>yarn.federation.subcluster-resolver.class</name>
|
||||
<value>org.apache.hadoop.yarn.server.federation.resolver.DefaultSubClusterResolverImpl</value>
|
||||
</property>
|
||||
|
||||
<property>
|
||||
<description>
|
||||
|
@ -21,8 +21,8 @@
|
||||
import org.apache.hadoop.yarn.exceptions.YarnException;
|
||||
import org.apache.hadoop.yarn.server.federation.store.records.SubClusterId;
|
||||
|
||||
import java.util.HashMap;
|
||||
import java.util.Set;
|
||||
import java.util.concurrent.ConcurrentHashMap;
|
||||
import java.util.Map;
|
||||
|
||||
/**
|
||||
@ -31,9 +31,9 @@
|
||||
*/
|
||||
public abstract class AbstractSubClusterResolver implements SubClusterResolver {
|
||||
private Map<String, SubClusterId> nodeToSubCluster =
|
||||
new HashMap<String, SubClusterId>();
|
||||
new ConcurrentHashMap<String, SubClusterId>();
|
||||
private Map<String, Set<SubClusterId>> rackToSubClusters =
|
||||
new HashMap<String, Set<SubClusterId>>();
|
||||
new ConcurrentHashMap<String, Set<SubClusterId>>();
|
||||
|
||||
@Override
|
||||
public SubClusterId getSubClusterForNode(String nodename)
|
||||
|
@ -25,8 +25,8 @@
|
||||
import org.apache.hadoop.yarn.server.federation.store.records.SubClusterId;
|
||||
|
||||
/**
|
||||
* An utility that helps to determine the sub-cluster that a specified node
|
||||
* belongs to.
|
||||
* An utility that helps to determine the sub-cluster that a specified node or
|
||||
* rack belongs to. All implementing classes should be thread-safe.
|
||||
*/
|
||||
public interface SubClusterResolver extends Configurable {
|
||||
|
||||
|
@ -46,6 +46,7 @@
|
||||
import org.apache.hadoop.yarn.conf.YarnConfiguration;
|
||||
import org.apache.hadoop.yarn.exceptions.YarnException;
|
||||
import org.apache.hadoop.yarn.exceptions.YarnRuntimeException;
|
||||
import org.apache.hadoop.yarn.server.federation.resolver.SubClusterResolver;
|
||||
import org.apache.hadoop.yarn.server.federation.store.FederationStateStore;
|
||||
import org.apache.hadoop.yarn.server.federation.store.records.AddApplicationHomeSubClusterRequest;
|
||||
import org.apache.hadoop.yarn.server.federation.store.records.AddApplicationHomeSubClusterResponse;
|
||||
@ -90,6 +91,7 @@ public final class FederationStateStoreFacade {
|
||||
private int cacheTimeToLive;
|
||||
private Configuration conf;
|
||||
private Cache<Object, Object> cache;
|
||||
private SubClusterResolver subclusterResolver;
|
||||
|
||||
private FederationStateStoreFacade() {
|
||||
initializeFacadeInternal(new Configuration());
|
||||
@ -104,6 +106,12 @@ private void initializeFacadeInternal(Configuration config) {
|
||||
FederationStateStore.class, createRetryPolicy(conf));
|
||||
this.stateStore.init(conf);
|
||||
|
||||
this.subclusterResolver = createInstance(conf,
|
||||
YarnConfiguration.FEDERATION_CLUSTER_RESOLVER_CLASS,
|
||||
YarnConfiguration.DEFAULT_FEDERATION_CLUSTER_RESOLVER_CLASS,
|
||||
SubClusterResolver.class);
|
||||
this.subclusterResolver.load();
|
||||
|
||||
initCache();
|
||||
|
||||
} catch (YarnException ex) {
|
||||
@ -347,6 +355,15 @@ public SubClusterId getApplicationHomeSubCluster(ApplicationId appId)
|
||||
return response.getApplicationHomeSubCluster().getHomeSubCluster();
|
||||
}
|
||||
|
||||
/**
|
||||
* Get the singleton instance of SubClusterResolver.
|
||||
*
|
||||
* @return SubClusterResolver instance
|
||||
*/
|
||||
public SubClusterResolver getSubClusterResolver() {
|
||||
return this.subclusterResolver;
|
||||
}
|
||||
|
||||
/**
|
||||
* Helper method to create instances of Object using the class name defined in
|
||||
* the configuration object. The instances creates {@link RetryProxy} using
|
||||
@ -359,23 +376,40 @@ public SubClusterId getApplicationHomeSubCluster(ApplicationId appId)
|
||||
* @param retryPolicy the policy for retrying method call failures
|
||||
* @return a retry proxy for the specified interface
|
||||
*/
|
||||
@SuppressWarnings("unchecked")
|
||||
public static <T> Object createRetryInstance(Configuration conf,
|
||||
String configuredClassName, String defaultValue, Class<T> type,
|
||||
RetryPolicy retryPolicy) {
|
||||
|
||||
return RetryProxy.create(type,
|
||||
createInstance(conf, configuredClassName, defaultValue, type),
|
||||
retryPolicy);
|
||||
}
|
||||
|
||||
/**
|
||||
* Helper method to create instances of Object using the class name specified
|
||||
* in the configuration object.
|
||||
*
|
||||
* @param conf the yarn configuration
|
||||
* @param configuredClassName the configuration provider key
|
||||
* @param defaultValue the default implementation class
|
||||
* @param type the required interface/base class
|
||||
* @param <T> The type of the instance to create
|
||||
* @return the instances created
|
||||
*/
|
||||
@SuppressWarnings("unchecked")
|
||||
public static <T> T createInstance(Configuration conf,
|
||||
String configuredClassName, String defaultValue, Class<T> type) {
|
||||
|
||||
String className = conf.get(configuredClassName, defaultValue);
|
||||
try {
|
||||
Class<?> clusterResolverClass = conf.getClassByName(className);
|
||||
if (type.isAssignableFrom(clusterResolverClass)) {
|
||||
return RetryProxy.create(type,
|
||||
(T) ReflectionUtils.newInstance(clusterResolverClass, conf),
|
||||
retryPolicy);
|
||||
return (T) ReflectionUtils.newInstance(clusterResolverClass, conf);
|
||||
} else {
|
||||
throw new YarnRuntimeException(
|
||||
"Class: " + className + " not instance of " + type.getSimpleName());
|
||||
throw new YarnRuntimeException("Class: " + className
|
||||
+ " not instance of " + type.getCanonicalName());
|
||||
}
|
||||
} catch (Exception e) {
|
||||
} catch (ClassNotFoundException e) {
|
||||
throw new YarnRuntimeException("Could not instantiate : " + className, e);
|
||||
}
|
||||
}
|
||||
|
Loading…
Reference in New Issue
Block a user