From a8554e63182de7b7c038e5d29fdd9ee93e3cb492 Mon Sep 17 00:00:00 2001 From: slfan1989 Date: Sun, 5 Nov 2023 13:52:23 +0800 Subject: [PATCH 1/4] YARN-11610. [Federation] Add WeightedHomePolicyManager. --- .../manager/WeightedHomePolicyManager.java | 67 +++++++++++++++++++ .../TestWeightedHomePolicyManager.java | 21 ++++++ 2 files changed, 88 insertions(+) create mode 100644 hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/federation/policies/manager/WeightedHomePolicyManager.java create mode 100644 hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/test/java/org/apache/hadoop/yarn/server/federation/policies/manager/TestWeightedHomePolicyManager.java 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/manager/WeightedHomePolicyManager.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/federation/policies/manager/WeightedHomePolicyManager.java new file mode 100644 index 0000000000000..f2bdb29554aef --- /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/manager/WeightedHomePolicyManager.java @@ -0,0 +1,67 @@ +/** + * 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.manager; + +import org.apache.hadoop.classification.VisibleForTesting; +import org.apache.hadoop.yarn.server.federation.policies.amrmproxy.HomeAMRMProxyPolicy; +import org.apache.hadoop.yarn.server.federation.policies.amrmproxy.LocalityMulticastAMRMProxyPolicy; +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.policies.router.LocalityRouterPolicy; +import org.apache.hadoop.yarn.server.federation.policies.router.WeightedRandomRouterPolicy; +import org.apache.hadoop.yarn.server.federation.store.records.SubClusterPolicyConfiguration; + +import java.nio.ByteBuffer; + +/** + * Policy that allows operator to configure "weights" for routing. This picks a + * {@link WeightedRandomRouterPolicy} for the router and a {@link + * HomeAMRMProxyPolicy} for the amrmproxy as they are designed to + * work together. + */ +public class WeightedHomePolicyManager extends AbstractPolicyManager { + + private WeightedPolicyInfo weightedPolicyInfo; + + public WeightedHomePolicyManager() { + // this structurally hard-codes two compatible policies for Router and + // AMRMProxy. + routerFederationPolicy = WeightedRandomRouterPolicy.class; + amrmProxyFederationPolicy = HomeAMRMProxyPolicy.class; + weightedPolicyInfo = new WeightedPolicyInfo(); + } + + @Override + public SubClusterPolicyConfiguration serializeConf() + throws FederationPolicyInitializationException { + ByteBuffer buf = weightedPolicyInfo.toByteBuffer(); + return SubClusterPolicyConfiguration + .newInstance(getQueue(), this.getClass().getCanonicalName(), buf); + } + + @VisibleForTesting + public WeightedPolicyInfo getWeightedPolicyInfo() { + return weightedPolicyInfo; + } + + @VisibleForTesting + public void setWeightedPolicyInfo( + WeightedPolicyInfo weightedPolicyInfo) { + this.weightedPolicyInfo = weightedPolicyInfo; + } +} 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/manager/TestWeightedHomePolicyManager.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/test/java/org/apache/hadoop/yarn/server/federation/policies/manager/TestWeightedHomePolicyManager.java new file mode 100644 index 0000000000000..a506d517781f8 --- /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/manager/TestWeightedHomePolicyManager.java @@ -0,0 +1,21 @@ +/** + * 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.manager; + +public class TestWeightedHomePolicyManager extends BasePolicyManagerTest { + +} From c6561757110796b846714b7b2517c4345dbe7a69 Mon Sep 17 00:00:00 2001 From: slfan1989 Date: Wed, 8 Nov 2023 21:03:05 +0800 Subject: [PATCH 2/4] YARN-11610. Fix CheckStyle. --- .../TestWeightedHomePolicyManager.java | 41 +++++++++++++++++++ 1 file changed, 41 insertions(+) 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/manager/TestWeightedHomePolicyManager.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/test/java/org/apache/hadoop/yarn/server/federation/policies/manager/TestWeightedHomePolicyManager.java index a506d517781f8..c7cb8b7de002a 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/test/java/org/apache/hadoop/yarn/server/federation/policies/manager/TestWeightedHomePolicyManager.java +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/test/java/org/apache/hadoop/yarn/server/federation/policies/manager/TestWeightedHomePolicyManager.java @@ -16,6 +16,47 @@ */ package org.apache.hadoop.yarn.server.federation.policies.manager; +import org.apache.hadoop.yarn.server.federation.policies.amrmproxy.HomeAMRMProxyPolicy; +import org.apache.hadoop.yarn.server.federation.policies.dao.WeightedPolicyInfo; +import org.apache.hadoop.yarn.server.federation.policies.router.WeightedRandomRouterPolicy; +import org.apache.hadoop.yarn.server.federation.store.records.SubClusterId; +import org.apache.hadoop.yarn.server.federation.store.records.SubClusterIdInfo; +import org.junit.Assert; +import org.junit.Before; +import org.junit.Test; + +import java.util.HashMap; +import java.util.Map; + public class TestWeightedHomePolicyManager extends BasePolicyManagerTest { + private WeightedPolicyInfo policyInfo; + + @Before + public void setup() { + // configure a policy + wfp = new WeightedHomePolicyManager(); + wfp.setQueue("queue1"); + SubClusterId sc1 = SubClusterId.newInstance("sc1"); + policyInfo = new WeightedPolicyInfo(); + + Map routerWeights = new HashMap<>(); + routerWeights.put(new SubClusterIdInfo(sc1), 0.2f); + policyInfo.setRouterPolicyWeights(routerWeights); + + ((WeightedHomePolicyManager) wfp).setWeightedPolicyInfo(policyInfo); + + //set expected params that the base test class will use for tests + expectedPolicyManager = WeightedHomePolicyManager.class; + expectedAMRMProxyPolicy = HomeAMRMProxyPolicy.class; + expectedRouterPolicy = WeightedRandomRouterPolicy.class; + } + @Test + public void testPolicyInfoSetCorrectly() throws Exception { + serializeAndDeserializePolicyManager(wfp, expectedPolicyManager, + expectedAMRMProxyPolicy, expectedRouterPolicy); + // check the policyInfo propagates through ser/der correctly + Assert.assertEquals(((WeightedHomePolicyManager) wfp) + .getWeightedPolicyInfo(), policyInfo); + } } From 7f1095497b51c23bef48941f9bc3137069a35d0b Mon Sep 17 00:00:00 2001 From: slfan1989 Date: Thu, 9 Nov 2023 23:18:28 +0800 Subject: [PATCH 3/4] YARN-11610. Fix CheckStyle. --- .../policies/manager/WeightedHomePolicyManager.java | 2 -- .../policies/manager/TestWeightedHomePolicyManager.java | 8 ++++---- 2 files changed, 4 insertions(+), 6 deletions(-) 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/manager/WeightedHomePolicyManager.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/federation/policies/manager/WeightedHomePolicyManager.java index f2bdb29554aef..370594ec32d36 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/federation/policies/manager/WeightedHomePolicyManager.java +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/federation/policies/manager/WeightedHomePolicyManager.java @@ -19,10 +19,8 @@ import org.apache.hadoop.classification.VisibleForTesting; import org.apache.hadoop.yarn.server.federation.policies.amrmproxy.HomeAMRMProxyPolicy; -import org.apache.hadoop.yarn.server.federation.policies.amrmproxy.LocalityMulticastAMRMProxyPolicy; 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.policies.router.LocalityRouterPolicy; import org.apache.hadoop.yarn.server.federation.policies.router.WeightedRandomRouterPolicy; import org.apache.hadoop.yarn.server.federation.store.records.SubClusterPolicyConfiguration; 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/manager/TestWeightedHomePolicyManager.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/test/java/org/apache/hadoop/yarn/server/federation/policies/manager/TestWeightedHomePolicyManager.java index c7cb8b7de002a..d41db74cbf6f1 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/test/java/org/apache/hadoop/yarn/server/federation/policies/manager/TestWeightedHomePolicyManager.java +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/test/java/org/apache/hadoop/yarn/server/federation/policies/manager/TestWeightedHomePolicyManager.java @@ -45,10 +45,10 @@ public void setup() { ((WeightedHomePolicyManager) wfp).setWeightedPolicyInfo(policyInfo); - //set expected params that the base test class will use for tests - expectedPolicyManager = WeightedHomePolicyManager.class; - expectedAMRMProxyPolicy = HomeAMRMProxyPolicy.class; - expectedRouterPolicy = WeightedRandomRouterPolicy.class; + //set expected params that the base test class will use for tests + expectedPolicyManager = WeightedHomePolicyManager.class; + expectedAMRMProxyPolicy = HomeAMRMProxyPolicy.class; + expectedRouterPolicy = WeightedRandomRouterPolicy.class; } @Test From 9dbdadf8c8ff9c442512fe5037c5cdce0f8f2286 Mon Sep 17 00:00:00 2001 From: slfan1989 Date: Thu, 16 Nov 2023 22:47:00 +0800 Subject: [PATCH 4/4] YARN-11610. Fix CheckStyle. --- .../policies/manager/TestWeightedHomePolicyManager.java | 9 +++++---- 1 file changed, 5 insertions(+), 4 deletions(-) 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/manager/TestWeightedHomePolicyManager.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/test/java/org/apache/hadoop/yarn/server/federation/policies/manager/TestWeightedHomePolicyManager.java index d41db74cbf6f1..3d0c5895f42d0 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/test/java/org/apache/hadoop/yarn/server/federation/policies/manager/TestWeightedHomePolicyManager.java +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/test/java/org/apache/hadoop/yarn/server/federation/policies/manager/TestWeightedHomePolicyManager.java @@ -34,16 +34,17 @@ public class TestWeightedHomePolicyManager extends BasePolicyManagerTest { @Before public void setup() { // configure a policy - wfp = new WeightedHomePolicyManager(); - wfp.setQueue("queue1"); + WeightedHomePolicyManager whpm = new WeightedHomePolicyManager(); + whpm.setQueue("queue1"); + SubClusterId sc1 = SubClusterId.newInstance("sc1"); policyInfo = new WeightedPolicyInfo(); - Map routerWeights = new HashMap<>(); routerWeights.put(new SubClusterIdInfo(sc1), 0.2f); policyInfo.setRouterPolicyWeights(routerWeights); - ((WeightedHomePolicyManager) wfp).setWeightedPolicyInfo(policyInfo); + whpm.setWeightedPolicyInfo(policyInfo); + this.wfp = whpm; //set expected params that the base test class will use for tests expectedPolicyManager = WeightedHomePolicyManager.class;