() {
@Override
- public Void run() throws IOException, YarnException {
- ((MutableConfScheduler) scheduler).updateConfiguration(callerUGI,
- mutationInfo);
+ public Void run() throws Exception {
+ MutableConfigurationProvider provider = ((MutableConfScheduler)
+ scheduler).getMutableConfProvider();
+ if (!provider.getAclMutationPolicy().isMutationAllowed(callerUGI,
+ mutationInfo)) {
+ throw new org.apache.hadoop.security.AccessControlException("User"
+ + " is not admin of all modified queues.");
+ }
+ provider.logAndApplyMutation(callerUGI, mutationInfo);
+ try {
+ rm.getRMContext().getRMAdminService().refreshQueues();
+ } catch (IOException | YarnException e) {
+ provider.confirmPendingMutation(false);
+ throw e;
+ }
+ provider.confirmPendingMutation(true);
return null;
}
});
} catch (IOException e) {
+ LOG.error("Exception thrown when modifying configuration.", e);
return Response.status(Status.BAD_REQUEST).entity(e.getMessage())
.build();
}
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/conf/ConfigurationStoreBaseTest.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/conf/ConfigurationStoreBaseTest.java
new file mode 100644
index 00000000000..bbe95700f18
--- /dev/null
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/conf/ConfigurationStoreBaseTest.java
@@ -0,0 +1,90 @@
+/**
+ * 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.resourcemanager.scheduler.capacity.conf;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.yarn.server.resourcemanager.RMContext;
+import org.junit.Before;
+import org.junit.Test;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNull;
+
+/**
+ * Base class for {@link YarnConfigurationStore} implementations.
+ */
+public abstract class ConfigurationStoreBaseTest {
+
+ protected YarnConfigurationStore confStore = createConfStore();
+
+ protected abstract YarnConfigurationStore createConfStore();
+
+ protected Configuration conf;
+ protected Configuration schedConf;
+ protected RMContext rmContext;
+
+ protected static final String TEST_USER = "testUser";
+
+ @Before
+ public void setUp() throws Exception {
+ this.conf = new Configuration();
+ this.schedConf = new Configuration(false);
+ }
+
+ @Test
+ public void testConfigurationUpdate() throws Exception {
+ schedConf.set("key1", "val1");
+ confStore.initialize(conf, schedConf, rmContext);
+ assertEquals("val1", confStore.retrieve().get("key1"));
+
+ Map update1 = new HashMap<>();
+ update1.put("keyUpdate1", "valUpdate1");
+ YarnConfigurationStore.LogMutation mutation1 =
+ new YarnConfigurationStore.LogMutation(update1, TEST_USER);
+ confStore.logMutation(mutation1);
+ confStore.confirmMutation(true);
+ assertEquals("valUpdate1", confStore.retrieve().get("keyUpdate1"));
+
+ Map update2 = new HashMap<>();
+ update2.put("keyUpdate2", "valUpdate2");
+ YarnConfigurationStore.LogMutation mutation2 =
+ new YarnConfigurationStore.LogMutation(update2, TEST_USER);
+ confStore.logMutation(mutation2);
+ confStore.confirmMutation(false);
+ assertNull("Configuration should not be updated",
+ confStore.retrieve().get("keyUpdate2"));
+ }
+
+ @Test
+ public void testNullConfigurationUpdate() throws Exception {
+ schedConf.set("key", "val");
+ confStore.initialize(conf, schedConf, rmContext);
+ assertEquals("val", confStore.retrieve().get("key"));
+
+ Map update = new HashMap<>();
+ update.put("key", null);
+ YarnConfigurationStore.LogMutation mutation =
+ new YarnConfigurationStore.LogMutation(update, TEST_USER);
+ confStore.logMutation(mutation);
+ confStore.confirmMutation(true);
+ assertNull(confStore.retrieve().get("key"));
+ }
+}
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/conf/TestInMemoryConfigurationStore.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/conf/TestInMemoryConfigurationStore.java
new file mode 100644
index 00000000000..c40d16a27cb
--- /dev/null
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/conf/TestInMemoryConfigurationStore.java
@@ -0,0 +1,30 @@
+/**
+ * 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.resourcemanager.scheduler.capacity.conf;
+
+/**
+ * Tests {@link InMemoryConfigurationStore}.
+ */
+public class TestInMemoryConfigurationStore extends ConfigurationStoreBaseTest {
+
+ @Override
+ protected YarnConfigurationStore createConfStore() {
+ return new InMemoryConfigurationStore();
+ }
+}
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/conf/TestMutableCSConfigurationProvider.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/conf/TestMutableCSConfigurationProvider.java
index 635a184e8ff..9b080cd1e2d 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/conf/TestMutableCSConfigurationProvider.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/conf/TestMutableCSConfigurationProvider.java
@@ -20,7 +20,6 @@ package org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.conf;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.security.UserGroupInformation;
-import org.apache.hadoop.yarn.exceptions.YarnException;
import org.apache.hadoop.yarn.server.resourcemanager.AdminService;
import org.apache.hadoop.yarn.server.resourcemanager.RMContext;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.CapacityScheduler;
@@ -30,14 +29,11 @@ import org.apache.hadoop.yarn.webapp.dao.SchedConfUpdateInfo;
import org.junit.Before;
import org.junit.Test;
-import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNull;
-import static org.mockito.Mockito.doNothing;
-import static org.mockito.Mockito.doThrow;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
@@ -82,25 +78,21 @@ public class TestMutableCSConfigurationProvider {
}
@Test
- public void testInMemoryBackedProvider() throws IOException, YarnException {
+ public void testInMemoryBackedProvider() throws Exception {
Configuration conf = new Configuration();
confProvider.init(conf);
assertNull(confProvider.loadConfiguration(conf)
.get("yarn.scheduler.capacity.root.a.goodKey"));
- doNothing().when(adminService).refreshQueues();
- confProvider.mutateConfiguration(TEST_USER, goodUpdate);
+ confProvider.logAndApplyMutation(TEST_USER, goodUpdate);
+ confProvider.confirmPendingMutation(true);
assertEquals("goodVal", confProvider.loadConfiguration(conf)
.get("yarn.scheduler.capacity.root.a.goodKey"));
assertNull(confProvider.loadConfiguration(conf).get(
"yarn.scheduler.capacity.root.a.badKey"));
- doThrow(new IOException()).when(adminService).refreshQueues();
- try {
- confProvider.mutateConfiguration(TEST_USER, badUpdate);
- } catch (IOException e) {
- // Expected exception.
- }
+ confProvider.logAndApplyMutation(TEST_USER, badUpdate);
+ confProvider.confirmPendingMutation(false);
assertNull(confProvider.loadConfiguration(conf).get(
"yarn.scheduler.capacity.root.a.badKey"));
}
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/conf/TestYarnConfigurationStore.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/conf/TestYarnConfigurationStore.java
deleted file mode 100644
index 631ce657e84..00000000000
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/conf/TestYarnConfigurationStore.java
+++ /dev/null
@@ -1,71 +0,0 @@
-/**
- * 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.resourcemanager.scheduler.capacity.conf;
-
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.conf.YarnConfigurationStore.LogMutation;
-import org.junit.Before;
-import org.junit.Test;
-
-import java.io.IOException;
-import java.util.HashMap;
-import java.util.Map;
-
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertNull;
-
-public class TestYarnConfigurationStore {
-
- private YarnConfigurationStore confStore;
- private Configuration schedConf;
-
- private static final String testUser = "testUser";
-
- @Before
- public void setUp() {
- schedConf = new Configuration(false);
- schedConf.set("key1", "val1");
- }
-
- @Test
- public void testInMemoryConfigurationStore() throws IOException {
- confStore = new InMemoryConfigurationStore();
- confStore.initialize(new Configuration(), schedConf);
- assertEquals("val1", confStore.retrieve().get("key1"));
-
- Map update1 = new HashMap<>();
- update1.put("keyUpdate1", "valUpdate1");
- LogMutation mutation1 = new LogMutation(update1, testUser);
- long id = confStore.logMutation(mutation1);
- assertEquals(1, confStore.getPendingMutations().size());
- confStore.confirmMutation(id, true);
- assertEquals("valUpdate1", confStore.retrieve().get("keyUpdate1"));
- assertEquals(0, confStore.getPendingMutations().size());
-
- Map update2 = new HashMap<>();
- update2.put("keyUpdate2", "valUpdate2");
- LogMutation mutation2 = new LogMutation(update2, testUser);
- id = confStore.logMutation(mutation2);
- assertEquals(1, confStore.getPendingMutations().size());
- confStore.confirmMutation(id, false);
- assertNull("Configuration should not be updated",
- confStore.retrieve().get("keyUpdate2"));
- assertEquals(0, confStore.getPendingMutations().size());
- }
-}
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/conf/TestZKConfigurationStore.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/conf/TestZKConfigurationStore.java
new file mode 100644
index 00000000000..3cfa8da8047
--- /dev/null
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/conf/TestZKConfigurationStore.java
@@ -0,0 +1,312 @@
+/**
+ * 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.resourcemanager.scheduler.capacity.conf;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.curator.framework.CuratorFramework;
+import org.apache.curator.framework.CuratorFrameworkFactory;
+import org.apache.curator.retry.RetryNTimes;
+import org.apache.curator.test.TestingServer;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.CommonConfigurationKeys;
+import org.apache.hadoop.ha.HAServiceProtocol;
+import org.apache.hadoop.security.UserGroupInformation;
+import org.apache.hadoop.service.Service;
+import org.apache.hadoop.yarn.conf.HAUtil;
+import org.apache.hadoop.yarn.conf.YarnConfiguration;
+import org.apache.hadoop.yarn.server.resourcemanager.MockRM;
+import org.apache.hadoop.yarn.server.resourcemanager.ResourceManager;
+import org.apache.hadoop.yarn.server.resourcemanager.recovery.ZKRMStateStore;
+import org.apache.hadoop.yarn.server.resourcemanager.scheduler.MutableConfScheduler;
+import org.apache.hadoop.yarn.server.resourcemanager.scheduler.MutableConfigurationProvider;
+import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.CapacityScheduler;
+import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.CapacitySchedulerConfiguration;
+import org.apache.hadoop.yarn.webapp.dao.SchedConfUpdateInfo;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.io.IOException;
+import java.util.HashMap;
+import java.util.LinkedList;
+import java.util.Map;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNull;
+
+/**
+ * Tests {@link ZKConfigurationStore}.
+ */
+public class TestZKConfigurationStore extends ConfigurationStoreBaseTest {
+
+ public static final Log LOG =
+ LogFactory.getLog(TestZKConfigurationStore.class);
+
+ private static final int ZK_TIMEOUT_MS = 10000;
+ private TestingServer curatorTestingServer;
+ private CuratorFramework curatorFramework;
+ private ResourceManager rm;
+
+ public static TestingServer setupCuratorServer() throws Exception {
+ TestingServer curatorTestingServer = new TestingServer();
+ curatorTestingServer.start();
+ return curatorTestingServer;
+ }
+
+ public static CuratorFramework setupCuratorFramework(
+ TestingServer curatorTestingServer) throws Exception {
+ CuratorFramework curatorFramework = CuratorFrameworkFactory.builder()
+ .connectString(curatorTestingServer.getConnectString())
+ .retryPolicy(new RetryNTimes(100, 100))
+ .build();
+ curatorFramework.start();
+ return curatorFramework;
+ }
+
+ @Before
+ public void setUp() throws Exception {
+ super.setUp();
+ curatorTestingServer = setupCuratorServer();
+ curatorFramework = setupCuratorFramework(curatorTestingServer);
+
+ conf.set(CommonConfigurationKeys.ZK_ADDRESS,
+ curatorTestingServer.getConnectString());
+ rm = new MockRM(conf);
+ rm.start();
+ rmContext = rm.getRMContext();
+ }
+
+ @After
+ public void cleanup() throws IOException {
+ rm.stop();
+ curatorFramework.close();
+ curatorTestingServer.stop();
+ }
+
+ @Test
+ public void testVersioning() throws Exception {
+ confStore.initialize(conf, schedConf, rmContext);
+ assertNull(confStore.getConfStoreVersion());
+ confStore.checkVersion();
+ assertEquals(ZKConfigurationStore.CURRENT_VERSION_INFO,
+ confStore.getConfStoreVersion());
+ }
+
+ @Test
+ public void testPersistConfiguration() throws Exception {
+ schedConf.set("key", "val");
+ confStore.initialize(conf, schedConf, rmContext);
+ assertEquals("val", confStore.retrieve().get("key"));
+
+ // Create a new configuration store, and check for old configuration
+ confStore = createConfStore();
+ schedConf.set("key", "badVal");
+ // Should ignore passed-in scheduler configuration.
+ confStore.initialize(conf, schedConf, rmContext);
+ assertEquals("val", confStore.retrieve().get("key"));
+ }
+
+
+ @Test
+ public void testPersistUpdatedConfiguration() throws Exception {
+ confStore.initialize(conf, schedConf, rmContext);
+ assertNull(confStore.retrieve().get("key"));
+
+ Map update = new HashMap<>();
+ update.put("key", "val");
+ YarnConfigurationStore.LogMutation mutation =
+ new YarnConfigurationStore.LogMutation(update, TEST_USER);
+ confStore.logMutation(mutation);
+ confStore.confirmMutation(true);
+ assertEquals("val", confStore.retrieve().get("key"));
+
+ // Create a new configuration store, and check for updated configuration
+ confStore = createConfStore();
+ schedConf.set("key", "badVal");
+ // Should ignore passed-in scheduler configuration.
+ confStore.initialize(conf, schedConf, rmContext);
+ assertEquals("val", confStore.retrieve().get("key"));
+ }
+
+ @Test
+ public void testMaxLogs() throws Exception {
+ conf.setLong(YarnConfiguration.RM_SCHEDCONF_MAX_LOGS, 2);
+ confStore.initialize(conf, schedConf, rmContext);
+ LinkedList logs =
+ ((ZKConfigurationStore) confStore).getLogs();
+ assertEquals(0, logs.size());
+
+ Map update1 = new HashMap<>();
+ update1.put("key1", "val1");
+ YarnConfigurationStore.LogMutation mutation =
+ new YarnConfigurationStore.LogMutation(update1, TEST_USER);
+ confStore.logMutation(mutation);
+ logs = ((ZKConfigurationStore) confStore).getLogs();
+ assertEquals(1, logs.size());
+ assertEquals("val1", logs.get(0).getUpdates().get("key1"));
+ confStore.confirmMutation(true);
+ assertEquals(1, logs.size());
+ assertEquals("val1", logs.get(0).getUpdates().get("key1"));
+
+ Map update2 = new HashMap<>();
+ update2.put("key2", "val2");
+ mutation = new YarnConfigurationStore.LogMutation(update2, TEST_USER);
+ confStore.logMutation(mutation);
+ logs = ((ZKConfigurationStore) confStore).getLogs();
+ assertEquals(2, logs.size());
+ assertEquals("val1", logs.get(0).getUpdates().get("key1"));
+ assertEquals("val2", logs.get(1).getUpdates().get("key2"));
+ confStore.confirmMutation(true);
+ assertEquals(2, logs.size());
+ assertEquals("val1", logs.get(0).getUpdates().get("key1"));
+ assertEquals("val2", logs.get(1).getUpdates().get("key2"));
+
+ // Next update should purge first update from logs.
+ Map update3 = new HashMap<>();
+ update3.put("key3", "val3");
+ mutation = new YarnConfigurationStore.LogMutation(update3, TEST_USER);
+ confStore.logMutation(mutation);
+ logs = ((ZKConfigurationStore) confStore).getLogs();
+ assertEquals(2, logs.size());
+ assertEquals("val2", logs.get(0).getUpdates().get("key2"));
+ assertEquals("val3", logs.get(1).getUpdates().get("key3"));
+ confStore.confirmMutation(true);
+ assertEquals(2, logs.size());
+ assertEquals("val2", logs.get(0).getUpdates().get("key2"));
+ assertEquals("val3", logs.get(1).getUpdates().get("key3"));
+ }
+
+ public Configuration createRMHAConf(String rmIds, String rmId,
+ int adminPort) {
+ Configuration conf = new YarnConfiguration();
+ conf.setBoolean(YarnConfiguration.RM_HA_ENABLED, true);
+ conf.set(YarnConfiguration.RM_HA_IDS, rmIds);
+ conf.setBoolean(YarnConfiguration.RECOVERY_ENABLED, true);
+ conf.set(CapacitySchedulerConfiguration.CS_CONF_PROVIDER,
+ CapacitySchedulerConfiguration.STORE_CS_CONF_PROVIDER);
+ conf.set(YarnConfiguration.SCHEDULER_CONFIGURATION_STORE_CLASS,
+ YarnConfiguration.ZK_CONFIGURATION_STORE);
+ conf.set(YarnConfiguration.RM_STORE, ZKRMStateStore.class.getName());
+ conf.set(YarnConfiguration.RM_ZK_ADDRESS,
+ curatorTestingServer.getConnectString());
+ conf.set(YarnConfiguration.RM_HA_ID, rmId);
+ conf.set(YarnConfiguration.RM_WEBAPP_ADDRESS, "localhost:0");
+ conf.setBoolean(YarnConfiguration.AUTO_FAILOVER_ENABLED, false);
+ for (String rpcAddress :
+ YarnConfiguration.getServiceAddressConfKeys(conf)) {
+ for (String id : HAUtil.getRMHAIds(conf)) {
+ conf.set(HAUtil.addSuffix(rpcAddress, id), "localhost:0");
+ }
+ }
+ conf.set(HAUtil.addSuffix(YarnConfiguration.RM_ADMIN_ADDRESS, rmId),
+ "localhost:" + adminPort);
+ return conf;
+ }
+
+ /**
+ * When failing over, new active RM should read from current state of store,
+ * including any updates when the new active RM was in standby.
+ * @throws Exception
+ */
+ @Test
+ public void testFailoverReadsFromUpdatedStore() throws Exception {
+ HAServiceProtocol.StateChangeRequestInfo req =
+ new HAServiceProtocol.StateChangeRequestInfo(
+ HAServiceProtocol.RequestSource.REQUEST_BY_USER);
+
+ Configuration conf1 = createRMHAConf("rm1,rm2", "rm1", 1234);
+ ResourceManager rm1 = new MockRM(conf1);
+ rm1.start();
+ rm1.getRMContext().getRMAdminService().transitionToActive(req);
+ assertEquals("RM with ZKStore didn't start",
+ Service.STATE.STARTED, rm1.getServiceState());
+ assertEquals("RM should be Active",
+ HAServiceProtocol.HAServiceState.ACTIVE,
+ rm1.getRMContext().getRMAdminService().getServiceStatus().getState());
+ assertNull(((MutableConfScheduler) rm1.getResourceScheduler())
+ .getConfiguration().get("key"));
+
+ Configuration conf2 = createRMHAConf("rm1,rm2", "rm2", 5678);
+ ResourceManager rm2 = new MockRM(conf2);
+ rm2.start();
+ assertEquals("RM should be Standby",
+ HAServiceProtocol.HAServiceState.STANDBY,
+ rm2.getRMContext().getRMAdminService().getServiceStatus().getState());
+
+ // Update configuration on RM1
+ SchedConfUpdateInfo schedConfUpdateInfo = new SchedConfUpdateInfo();
+ schedConfUpdateInfo.getGlobalParams().put("key", "val");
+ MutableConfigurationProvider confProvider = ((MutableConfScheduler)
+ rm1.getResourceScheduler()).getMutableConfProvider();
+ UserGroupInformation user = UserGroupInformation
+ .createUserForTesting(TEST_USER, new String[0]);
+ confProvider.logAndApplyMutation(user, schedConfUpdateInfo);
+ rm1.getResourceScheduler().reinitialize(conf1, rm1.getRMContext());
+ assertEquals("val", ((MutableConfScheduler) rm1.getResourceScheduler())
+ .getConfiguration().get("key"));
+ confProvider.confirmPendingMutation(true);
+ assertEquals("val", ((MutableCSConfigurationProvider) confProvider)
+ .getConfStore().retrieve().get("key"));
+ // Next update is not persisted, it should not be recovered
+ schedConfUpdateInfo.getGlobalParams().put("key", "badVal");
+ confProvider.logAndApplyMutation(user, schedConfUpdateInfo);
+
+ // Start RM2 and verifies it starts with updated configuration
+ rm2.getRMContext().getRMAdminService().transitionToActive(req);
+ assertEquals("RM with ZKStore didn't start",
+ Service.STATE.STARTED, rm2.getServiceState());
+ assertEquals("RM should be Active",
+ HAServiceProtocol.HAServiceState.ACTIVE,
+ rm2.getRMContext().getRMAdminService().getServiceStatus().getState());
+
+ for (int i = 0; i < ZK_TIMEOUT_MS / 50; i++) {
+ if (HAServiceProtocol.HAServiceState.ACTIVE ==
+ rm1.getRMContext().getRMAdminService().getServiceStatus()
+ .getState()) {
+ Thread.sleep(100);
+ }
+ }
+ assertEquals("RM should have been fenced",
+ HAServiceProtocol.HAServiceState.STANDBY,
+ rm1.getRMContext().getRMAdminService().getServiceStatus().getState());
+ assertEquals("RM should be Active",
+ HAServiceProtocol.HAServiceState.ACTIVE,
+ rm2.getRMContext().getRMAdminService().getServiceStatus().getState());
+
+ assertEquals("val", ((MutableCSConfigurationProvider) (
+ (CapacityScheduler) rm2.getResourceScheduler())
+ .getMutableConfProvider()).getConfStore().retrieve().get("key"));
+ assertEquals("val", ((MutableConfScheduler) rm2.getResourceScheduler())
+ .getConfiguration().get("key"));
+ // Transition to standby will set RM's HA status and then reinitialize in
+ // a separate thread. Despite asserting for STANDBY state, it's
+ // possible for reinitialization to be unfinished. Wait here for it to
+ // finish, otherwise closing rm1 will close zkManager and the unfinished
+ // reinitialization will throw an exception.
+ Thread.sleep(10000);
+ rm1.close();
+ rm2.close();
+ }
+
+ @Override
+ public YarnConfigurationStore createConfStore() {
+ return new ZKConfigurationStore();
+ }
+}