diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/federation/policies/amrmproxy/AbstractAMRMProxyPolicy.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/federation/policies/amrmproxy/AbstractAMRMProxyPolicy.java index e853744e106..07cd6db1909 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/federation/policies/amrmproxy/AbstractAMRMProxyPolicy.java +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/federation/policies/amrmproxy/AbstractAMRMProxyPolicy.java @@ -20,9 +20,12 @@ import java.util.Map; +import org.apache.hadoop.yarn.api.protocolrecords.AllocateResponse; +import org.apache.hadoop.yarn.exceptions.YarnException; import org.apache.hadoop.yarn.server.federation.policies.AbstractConfigurableFederationPolicy; import org.apache.hadoop.yarn.server.federation.policies.dao.WeightedPolicyInfo; import org.apache.hadoop.yarn.server.federation.policies.exceptions.FederationPolicyInitializationException; +import org.apache.hadoop.yarn.server.federation.store.records.SubClusterId; import org.apache.hadoop.yarn.server.federation.store.records.SubClusterIdInfo; /** @@ -44,4 +47,9 @@ public void validate(WeightedPolicyInfo newPolicyInfo) } } + @Override + public void notifyOfResponse(SubClusterId subClusterId, + AllocateResponse response) throws YarnException { + // By default, a stateless policy does not care about responses + } } diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/federation/policies/amrmproxy/LocalAMRMProxyPolicy.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/federation/policies/amrmproxy/LocalAMRMProxyPolicy.java new file mode 100644 index 00000000000..ef16e9f3a8f --- /dev/null +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/federation/policies/amrmproxy/LocalAMRMProxyPolicy.java @@ -0,0 +1,68 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.hadoop.yarn.server.federation.policies.amrmproxy; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.Map; + +import org.apache.hadoop.yarn.api.records.ResourceRequest; +import org.apache.hadoop.yarn.exceptions.YarnException; +import org.apache.hadoop.yarn.server.federation.policies.FederationPolicyInitializationContext; +import org.apache.hadoop.yarn.server.federation.policies.FederationPolicyInitializationContextValidator; +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.store.records.SubClusterId; + +/** + * An implementation of the {@link FederationAMRMProxyPolicy} that simply + * sends the {@link ResourceRequest} to the home subcluster. + */ +public class LocalAMRMProxyPolicy extends AbstractAMRMProxyPolicy { + + /** Identifier of the local subcluster. */ + private SubClusterId localSubcluster; + + @Override + public void reinitialize( + FederationPolicyInitializationContext policyContext) + throws FederationPolicyInitializationException { + + FederationPolicyInitializationContextValidator + .validate(policyContext, this.getClass().getCanonicalName()); + setPolicyContext(policyContext); + + this.localSubcluster = policyContext.getHomeSubcluster(); + } + + @Override + public Map> splitResourceRequests( + List resourceRequests) throws YarnException { + + getActiveSubclusters(); + if (localSubcluster == null) { + throw new FederationPolicyException("No local subcluster available"); + } + + List resourceRequestsCopy = + new ArrayList<>(resourceRequests); + return Collections.singletonMap(localSubcluster, resourceRequestsCopy); + } +} diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/test/java/org/apache/hadoop/yarn/server/federation/policies/amrmproxy/TestLocalAMRMProxyPolicy.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/test/java/org/apache/hadoop/yarn/server/federation/policies/amrmproxy/TestLocalAMRMProxyPolicy.java new file mode 100644 index 00000000000..daa40f9a1ae --- /dev/null +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/test/java/org/apache/hadoop/yarn/server/federation/policies/amrmproxy/TestLocalAMRMProxyPolicy.java @@ -0,0 +1,81 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.hadoop.yarn.server.federation.policies.amrmproxy; + +import static org.apache.hadoop.yarn.server.federation.utils.FederationPoliciesTestUtil.createResourceRequests; +import static org.apache.hadoop.yarn.server.federation.utils.FederationPoliciesTestUtil.initializePolicyContext; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +import java.util.List; +import java.util.Map; + +import org.apache.hadoop.yarn.api.records.ResourceRequest; +import org.apache.hadoop.yarn.server.federation.policies.BaseFederationPoliciesTest; +import org.apache.hadoop.yarn.server.federation.policies.dao.WeightedPolicyInfo; +import org.apache.hadoop.yarn.server.federation.store.records.SubClusterId; +import org.apache.hadoop.yarn.server.federation.store.records.SubClusterIdInfo; +import org.apache.hadoop.yarn.server.federation.store.records.SubClusterInfo; +import org.apache.hadoop.yarn.server.federation.store.records.SubClusterState; +import org.junit.Before; +import org.junit.Test; + +/** + * Simple test class for the {@link LocalAMRMProxyPolicy}. + */ +public class TestLocalAMRMProxyPolicy extends BaseFederationPoliciesTest { + + private static final int NUM_SUBCLUSTERS = 4; + + @Before + public void setUp() throws Exception { + setPolicy(new LocalAMRMProxyPolicy()); + // needed for base test to work + setPolicyInfo(mock(WeightedPolicyInfo.class)); + + for (int i = 0; i < NUM_SUBCLUSTERS; i++) { + SubClusterIdInfo sc = new SubClusterIdInfo("sc" + i); + SubClusterInfo sci = mock(SubClusterInfo.class); + when(sci.getState()).thenReturn(SubClusterState.SC_RUNNING); + when(sci.getSubClusterId()).thenReturn(sc.toId()); + getActiveSubclusters().put(sc.toId(), sci); + } + + initializePolicyContext( + getPolicy(), mock(WeightedPolicyInfo.class), getActiveSubclusters(), "sc2"); + } + + @Test + public void testSplitAllocateRequest() throws Exception { + // verify the request only goes to the local subcluster + String[] hosts = new String[] {"host0", "host1", "host2", "host3"}; + List resourceRequests = createResourceRequests( + hosts, 2 * 1024, 2, 1, 3, null, false); + + FederationAMRMProxyPolicy federationPolicy = + (FederationAMRMProxyPolicy)getPolicy(); + Map> response = + federationPolicy.splitResourceRequests(resourceRequests); + assertEquals(1, response.size()); + assertNotNull(response.get(SubClusterId.newInstance("sc2"))); + assertEquals(9, response.get(SubClusterId.newInstance("sc2")).size()); + } +} diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/test/java/org/apache/hadoop/yarn/server/federation/utils/FederationPoliciesTestUtil.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/test/java/org/apache/hadoop/yarn/server/federation/utils/FederationPoliciesTestUtil.java index 4954197ecfe..ce7512cceee 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/test/java/org/apache/hadoop/yarn/server/federation/utils/FederationPoliciesTestUtil.java +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/test/java/org/apache/hadoop/yarn/server/federation/utils/FederationPoliciesTestUtil.java @@ -141,11 +141,21 @@ public static void initializePolicyContext( public static void initializePolicyContext( ConfigurableFederationPolicy policy, - WeightedPolicyInfo policyInfo, Map activeSubclusters) throws YarnException { + WeightedPolicyInfo policyInfo, + Map activeSubclusters) + throws YarnException { + initializePolicyContext( + policy, policyInfo, activeSubclusters, "homesubcluster"); + } + + public static void initializePolicyContext( + ConfigurableFederationPolicy policy, + WeightedPolicyInfo policyInfo, + Map activeSubclusters, + String subclusterId) throws YarnException { FederationPolicyInitializationContext context = new FederationPolicyInitializationContext(null, initResolver(), - initFacade(), SubClusterId.newInstance("homesubcluster")); + initFacade(), SubClusterId.newInstance(subclusterId)); initializePolicyContext(context, policy, policyInfo, activeSubclusters); }