|
|
@@ -0,0 +1,128 @@
|
|
|
+/**
|
|
|
+ * 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;
|
|
|
+
|
|
|
+import java.util.ArrayList;
|
|
|
+import java.util.HashMap;
|
|
|
+import java.util.List;
|
|
|
+
|
|
|
+import org.apache.hadoop.yarn.api.records.ApplicationAccessType;
|
|
|
+import org.apache.hadoop.yarn.api.records.Container;
|
|
|
+import org.apache.hadoop.yarn.api.records.ContainerId;
|
|
|
+import org.apache.hadoop.yarn.api.records.ContainerState;
|
|
|
+import org.apache.hadoop.yarn.api.records.ResourceRequest;
|
|
|
+import org.apache.hadoop.yarn.conf.YarnConfiguration;
|
|
|
+import org.apache.hadoop.yarn.server.resourcemanager.MockAM;
|
|
|
+import org.apache.hadoop.yarn.server.resourcemanager.MockNM;
|
|
|
+import org.apache.hadoop.yarn.server.resourcemanager.MockRM;
|
|
|
+import org.apache.hadoop.yarn.server.resourcemanager.ParameterizedSchedulerTestBase;
|
|
|
+import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMApp;
|
|
|
+import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.RMContainerState;
|
|
|
+import org.junit.Test;
|
|
|
+
|
|
|
+@SuppressWarnings("unchecked")
|
|
|
+public class TestAbstractYarnScheduler extends ParameterizedSchedulerTestBase {
|
|
|
+
|
|
|
+ public TestAbstractYarnScheduler(SchedulerType type) {
|
|
|
+ super(type);
|
|
|
+ }
|
|
|
+
|
|
|
+ @Test(timeout = 60000)
|
|
|
+ public void testResourceRequestRestoreWhenRMContainerIsAtAllocated()
|
|
|
+ throws Exception {
|
|
|
+ configureScheduler();
|
|
|
+ YarnConfiguration conf = getConf();
|
|
|
+ MockRM rm1 = new MockRM(conf);
|
|
|
+ try {
|
|
|
+ rm1.start();
|
|
|
+ RMApp app1 =
|
|
|
+ rm1.submitApp(200, "name", "user",
|
|
|
+ new HashMap<ApplicationAccessType, String>(), false, "default",
|
|
|
+ -1, null, "Test", false, true);
|
|
|
+ MockNM nm1 =
|
|
|
+ new MockNM("127.0.0.1:1234", 10240, rm1.getResourceTrackerService());
|
|
|
+ nm1.registerNode();
|
|
|
+
|
|
|
+ MockNM nm2 =
|
|
|
+ new MockNM("127.0.0.1:2351", 10240, rm1.getResourceTrackerService());
|
|
|
+ nm2.registerNode();
|
|
|
+
|
|
|
+ MockAM am1 = MockRM.launchAndRegisterAM(app1, rm1, nm1);
|
|
|
+
|
|
|
+ int NUM_CONTAINERS = 1;
|
|
|
+ // allocate NUM_CONTAINERS containers
|
|
|
+ am1.allocate("127.0.0.1", 1024, NUM_CONTAINERS,
|
|
|
+ new ArrayList<ContainerId>());
|
|
|
+ nm1.nodeHeartbeat(true);
|
|
|
+
|
|
|
+ // wait for containers to be allocated.
|
|
|
+ List<Container> containers =
|
|
|
+ am1.allocate(new ArrayList<ResourceRequest>(),
|
|
|
+ new ArrayList<ContainerId>()).getAllocatedContainers();
|
|
|
+ while (containers.size() != NUM_CONTAINERS) {
|
|
|
+ nm1.nodeHeartbeat(true);
|
|
|
+ containers.addAll(am1.allocate(new ArrayList<ResourceRequest>(),
|
|
|
+ new ArrayList<ContainerId>()).getAllocatedContainers());
|
|
|
+ Thread.sleep(200);
|
|
|
+ }
|
|
|
+
|
|
|
+ // launch the 2nd container, for testing running container transferred.
|
|
|
+ nm1.nodeHeartbeat(am1.getApplicationAttemptId(), 2,
|
|
|
+ ContainerState.RUNNING);
|
|
|
+ ContainerId containerId2 =
|
|
|
+ ContainerId.newContainerId(am1.getApplicationAttemptId(), 2);
|
|
|
+ rm1.waitForState(nm1, containerId2, RMContainerState.RUNNING);
|
|
|
+
|
|
|
+ // 3rd container is in Allocated state.
|
|
|
+ am1.allocate("127.0.0.1", 1024, NUM_CONTAINERS,
|
|
|
+ new ArrayList<ContainerId>());
|
|
|
+ nm2.nodeHeartbeat(true);
|
|
|
+ ContainerId containerId3 =
|
|
|
+ ContainerId.newContainerId(am1.getApplicationAttemptId(), 3);
|
|
|
+ rm1.waitForContainerAllocated(nm2, containerId3);
|
|
|
+ rm1.waitForState(nm2, containerId3, RMContainerState.ALLOCATED);
|
|
|
+
|
|
|
+ // NodeManager restart
|
|
|
+ nm2.registerNode();
|
|
|
+
|
|
|
+ // NM restart kills all allocated and running containers.
|
|
|
+ rm1.waitForState(nm2, containerId3, RMContainerState.KILLED);
|
|
|
+
|
|
|
+ // The killed RMContainer request should be restored. In successive
|
|
|
+ // nodeHeartBeats AM should be able to get container allocated.
|
|
|
+ containers =
|
|
|
+ am1.allocate(new ArrayList<ResourceRequest>(),
|
|
|
+ new ArrayList<ContainerId>()).getAllocatedContainers();
|
|
|
+ while (containers.size() != NUM_CONTAINERS) {
|
|
|
+ nm2.nodeHeartbeat(true);
|
|
|
+ containers.addAll(am1.allocate(new ArrayList<ResourceRequest>(),
|
|
|
+ new ArrayList<ContainerId>()).getAllocatedContainers());
|
|
|
+ Thread.sleep(200);
|
|
|
+ }
|
|
|
+
|
|
|
+ nm2.nodeHeartbeat(am1.getApplicationAttemptId(), 4,
|
|
|
+ ContainerState.RUNNING);
|
|
|
+ ContainerId containerId4 =
|
|
|
+ ContainerId.newContainerId(am1.getApplicationAttemptId(), 4);
|
|
|
+ rm1.waitForState(nm2, containerId4, RMContainerState.RUNNING);
|
|
|
+ } finally {
|
|
|
+ rm1.stop();
|
|
|
+ }
|
|
|
+ }
|
|
|
+}
|
Arun Suresh