hadoop-common-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From jh...@apache.org
Subject [16/43] hadoop git commit: YARN-7258. Add Node and Rack Hints to Opportunistic Scheduler. (Kartheek Muthyala via asuresh).
Date Mon, 09 Oct 2017 18:46:03 GMT
YARN-7258. Add Node and Rack Hints to Opportunistic Scheduler. (Kartheek Muthyala via asuresh).

(cherry picked from commit b733348dde18a242e6c9074c512116a8baf1d281)


Project: http://git-wip-us.apache.org/repos/asf/hadoop/repo
Commit: http://git-wip-us.apache.org/repos/asf/hadoop/commit/cb91bb3c
Tree: http://git-wip-us.apache.org/repos/asf/hadoop/tree/cb91bb3c
Diff: http://git-wip-us.apache.org/repos/asf/hadoop/diff/cb91bb3c

Branch: refs/heads/YARN-5734-branch-2
Commit: cb91bb3c960ad4c6b1dc330fe9d60f8778ecf08a
Parents: 5cb3bd3
Author: Arun Suresh <asuresh@apache.org>
Authored: Mon Oct 2 18:01:51 2017 -0700
Committer: Arun Suresh <asuresh@apache.org>
Committed: Thu Oct 5 15:22:20 2017 -0700

----------------------------------------------------------------------
 .../yarn/api/records/ResourceRequest.java       |  16 +
 .../api/impl/TestDistributedScheduling.java     | 651 ----------------
 .../TestOpportunisticContainerAllocation.java   | 781 -------------------
 ...TestOpportunisticContainerAllocationE2E.java | 781 +++++++++++++++++++
 .../server/api/protocolrecords/RemoteNode.java  |  35 +
 .../impl/pb/RemoteNodePBImpl.java               |  19 +
 .../OpportunisticContainerAllocator.java        | 334 ++++++--
 .../OpportunisticContainerContext.java          |  68 +-
 .../yarn_server_common_service_protos.proto     |   1 +
 .../TestOpportunisticContainerAllocator.java    | 599 ++++++++++++++
 ...pportunisticContainerAllocatorAMService.java |   8 +-
 ...pportunisticContainerAllocatorAMService.java |   2 +
 12 files changed, 1778 insertions(+), 1517 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/hadoop/blob/cb91bb3c/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/ResourceRequest.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/ResourceRequest.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/ResourceRequest.java
index 5bedc87..e9be6c3 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/ResourceRequest.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/ResourceRequest.java
@@ -223,6 +223,22 @@ public abstract class ResourceRequest implements Comparable<ResourceRequest> {
     }
 
     /**
+     * Set the <code>executionTypeRequest</code> of the request with 'ensure
+     * execution type' flag set to true.
+     * @see ResourceRequest#setExecutionTypeRequest(
+     * ExecutionTypeRequest)
+     * @param executionType <code>executionType</code> of the request.
+     * @return {@link ResourceRequestBuilder}
+     */
+    @Public
+    @Evolving
+    public ResourceRequestBuilder executionType(ExecutionType executionType) {
+      resourceRequest.setExecutionTypeRequest(
+          ExecutionTypeRequest.newInstance(executionType, true));
+      return this;
+    }
+
+    /**
      * Set the <code>allocationRequestId</code> of the request.
      * @see ResourceRequest#setAllocationRequestId(long)
      * @param allocationRequestId

http://git-wip-us.apache.org/repos/asf/hadoop/blob/cb91bb3c/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/test/java/org/apache/hadoop/yarn/client/api/impl/TestDistributedScheduling.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/test/java/org/apache/hadoop/yarn/client/api/impl/TestDistributedScheduling.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/test/java/org/apache/hadoop/yarn/client/api/impl/TestDistributedScheduling.java
deleted file mode 100644
index d69a73c..0000000
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/test/java/org/apache/hadoop/yarn/client/api/impl/TestDistributedScheduling.java
+++ /dev/null
@@ -1,651 +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
- * <p>
- * http://www.apache.org/licenses/LICENSE-2.0
- * <p>
- * 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.client.api.impl;
-
-import org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.net.NetUtils;
-import org.apache.hadoop.service.Service;
-import org.apache.hadoop.yarn.api.ApplicationMasterProtocol;
-import org.apache.hadoop.yarn.api.protocolrecords.AllocateRequest;
-import org.apache.hadoop.yarn.api.protocolrecords.AllocateResponse;
-import org.apache.hadoop.yarn.api.protocolrecords.FinishApplicationMasterRequest;
-import org.apache.hadoop.yarn.api.protocolrecords.RegisterApplicationMasterRequest;
-import org.apache.hadoop.yarn.api.protocolrecords.RegisterApplicationMasterResponse;
-
-import org.apache.hadoop.yarn.api.records.ApplicationAttemptId;
-import org.apache.hadoop.yarn.api.records.ApplicationId;
-import org.apache.hadoop.yarn.api.records.Container;
-import org.apache.hadoop.yarn.api.records.ExecutionTypeRequest;
-import org.apache.hadoop.yarn.api.records.ContainerId;
-import org.apache.hadoop.yarn.api.records.ContainerState;
-import org.apache.hadoop.yarn.api.records.ContainerStatus;
-import org.apache.hadoop.yarn.api.records.ExecutionType;
-import org.apache.hadoop.yarn.api.records.FinalApplicationStatus;
-import org.apache.hadoop.yarn.api.records.NMToken;
-import org.apache.hadoop.yarn.api.records.NodeReport;
-import org.apache.hadoop.yarn.api.records.NodeState;
-import org.apache.hadoop.yarn.api.records.Priority;
-import org.apache.hadoop.yarn.api.records.Resource;
-import org.apache.hadoop.yarn.api.records.ResourceRequest;
-import org.apache.hadoop.yarn.api.records.Token;
-import org.apache.hadoop.yarn.client.api.AMRMClient;
-import org.apache.hadoop.yarn.client.api.AMRMClient.ContainerRequest;
-import org.apache.hadoop.yarn.client.api.YarnClient;
-import org.apache.hadoop.yarn.conf.YarnConfiguration;
-import org.apache.hadoop.yarn.security.ContainerTokenIdentifier;
-import org.apache.hadoop.yarn.server.MiniYARNCluster;
-import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMApp;
-import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMAppState;
-import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.RMContainer;
-import org.apache.hadoop.yarn.server.resourcemanager.scheduler.ResourceScheduler;
-import org.apache.hadoop.yarn.server.utils.BuilderUtils;
-import org.junit.After;
-import org.junit.Assert;
-import org.junit.Before;
-import org.junit.Test;
-import org.mockito.invocation.InvocationOnMock;
-import org.mockito.stubbing.Answer;
-
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
-import java.util.TreeSet;
-
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertNotNull;
-import static org.junit.Assert.assertNull;
-import static org.junit.Assert.assertTrue;
-import static org.mockito.Matchers.any;
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.when;
-
-/**
- * Validates End2End Distributed Scheduling flow which includes the AM
- * specifying OPPORTUNISTIC containers in its resource requests,
- * the AMRMProxyService on the NM, the DistributedScheduler RequestInterceptor
- * on the NM and the DistributedSchedulingProtocol used by the framework to talk
- * to the OpportunisticContainerAllocatorAMService running on the RM.
- */
-public class TestDistributedScheduling extends BaseAMRMProxyE2ETest {
-
-  private static final Log LOG =
-      LogFactory.getLog(TestDistributedScheduling.class);
-
-  protected MiniYARNCluster cluster;
-  protected YarnClient rmClient;
-  protected ApplicationMasterProtocol client;
-  protected Configuration conf;
-  protected Configuration yarnConf;
-  protected ApplicationAttemptId attemptId;
-  protected ApplicationId appId;
-
-  @Before
-  public void doBefore() throws Exception {
-    cluster = new MiniYARNCluster("testDistributedSchedulingE2E", 1, 1, 1);
-
-    conf = new YarnConfiguration();
-    conf.setBoolean(YarnConfiguration.
-        OPPORTUNISTIC_CONTAINER_ALLOCATION_ENABLED, true);
-    conf.setBoolean(YarnConfiguration.DIST_SCHEDULING_ENABLED, true);
-    conf.setInt(YarnConfiguration.NM_OPPORTUNISTIC_CONTAINERS_MAX_QUEUE_LENGTH,
-        10);
-    cluster.init(conf);
-    cluster.start();
-    yarnConf = cluster.getConfig();
-
-    // the client has to connect to AMRMProxy
-    yarnConf.set(YarnConfiguration.RM_SCHEDULER_ADDRESS,
-        YarnConfiguration.DEFAULT_AMRM_PROXY_ADDRESS);
-    rmClient = YarnClient.createYarnClient();
-    rmClient.init(yarnConf);
-    rmClient.start();
-
-    // Submit application
-    attemptId = createApp(rmClient, cluster, conf);
-    appId = attemptId.getApplicationId();
-    client = createAMRMProtocol(rmClient, appId, cluster, yarnConf);
-  }
-
-  @After
-  public void doAfter() throws Exception {
-    if (client != null) {
-      try {
-        client.finishApplicationMaster(FinishApplicationMasterRequest
-            .newInstance(FinalApplicationStatus.SUCCEEDED, "success", null));
-        rmClient.killApplication(attemptId.getApplicationId());
-        attemptId = null;
-      } catch (Exception e) {
-      }
-    }
-    if (rmClient != null) {
-      try {
-        rmClient.stop();
-      } catch (Exception e) {
-      }
-    }
-    if (cluster != null) {
-      try {
-        cluster.stop();
-      } catch (Exception e) {
-      }
-    }
-  }
-
-
-  /**
-   * Validates if Allocate Requests containing only OPPORTUNISTIC container
-   * requests are satisfied instantly.
-   *
-   * @throws Exception
-   */
-  @Test(timeout = 60000)
-  public void testOpportunisticExecutionTypeRequestE2E() throws Exception {
-    LOG.info("testDistributedSchedulingE2E - Register");
-
-    RegisterApplicationMasterResponse responseRegister =
-        client.registerApplicationMaster(RegisterApplicationMasterRequest
-            .newInstance(NetUtils.getHostname(), 1024, ""));
-
-    Assert.assertNotNull(responseRegister);
-    Assert.assertNotNull(responseRegister.getQueue());
-    Assert.assertNotNull(responseRegister.getApplicationACLs());
-    Assert.assertNotNull(responseRegister.getClientToAMTokenMasterKey());
-    Assert
-        .assertNotNull(responseRegister.getContainersFromPreviousAttempts());
-    Assert.assertNotNull(responseRegister.getSchedulerResourceTypes());
-    Assert.assertNotNull(responseRegister.getMaximumResourceCapability());
-
-    // Wait until the RM has been updated and verify
-    Map<ApplicationId, RMApp> rmApps =
-        cluster.getResourceManager().getRMContext().getRMApps();
-    boolean rmUpdated = false;
-    for (int i=0; i<10 && !rmUpdated; i++) {
-      sleep(100);
-      RMApp rmApp = rmApps.get(appId);
-      if (rmApp.getState() == RMAppState.RUNNING) {
-        rmUpdated = true;
-      }
-    }
-    RMApp rmApp = rmApps.get(appId);
-    Assert.assertEquals(RMAppState.RUNNING, rmApp.getState());
-
-    LOG.info("testDistributedSchedulingE2E - Allocate");
-
-    AllocateRequest request =
-        createAllocateRequest(rmClient.getNodeReports(NodeState.RUNNING));
-
-    // Replace 'ANY' requests with OPPORTUNISTIC aks and remove
-    // everything else
-    List<ResourceRequest> newAskList = new ArrayList<>();
-    for (ResourceRequest rr : request.getAskList()) {
-      if (ResourceRequest.ANY.equals(rr.getResourceName())) {
-        ResourceRequest newRR = ResourceRequest.newInstance(rr
-                .getPriority(), rr.getResourceName(),
-            rr.getCapability(), rr.getNumContainers(), rr.getRelaxLocality(),
-            rr.getNodeLabelExpression(),
-            ExecutionTypeRequest.newInstance(
-                ExecutionType.OPPORTUNISTIC, true));
-        newAskList.add(newRR);
-      }
-    }
-    request.setAskList(newAskList);
-
-    AllocateResponse allocResponse = client.allocate(request);
-    Assert.assertNotNull(allocResponse);
-
-    // Ensure that all the requests are satisfied immediately
-    Assert.assertEquals(2, allocResponse.getAllocatedContainers().size());
-
-    // Verify that the allocated containers are OPPORTUNISTIC
-    for (Container allocatedContainer : allocResponse
-        .getAllocatedContainers()) {
-      ContainerTokenIdentifier containerTokenIdentifier = BuilderUtils
-          .newContainerTokenIdentifier(
-              allocatedContainer.getContainerToken());
-      Assert.assertEquals(ExecutionType.OPPORTUNISTIC,
-          containerTokenIdentifier.getExecutionType());
-    }
-
-    // Check that the RM sees OPPORTUNISTIC containers
-    ResourceScheduler scheduler = cluster.getResourceManager()
-        .getResourceScheduler();
-    for (Container allocatedContainer : allocResponse
-        .getAllocatedContainers()) {
-      ContainerId containerId = allocatedContainer.getId();
-      RMContainer rmContainer = scheduler.getRMContainer(containerId);
-      Assert.assertEquals(ExecutionType.OPPORTUNISTIC,
-          rmContainer.getExecutionType());
-    }
-
-    LOG.info("testDistributedSchedulingE2E - Finish");
-  }
-
-  /**
-   * Validates if Allocate Requests containing both GUARANTEED and OPPORTUNISTIC
-   * container requests works as expected.
-   *
-   * @throws Exception
-   */
-  @Test(timeout = 60000)
-  public void testMixedExecutionTypeRequestE2E() throws Exception {
-    LOG.info("testDistributedSchedulingE2E - Register");
-
-    RegisterApplicationMasterResponse responseRegister =
-        client.registerApplicationMaster(RegisterApplicationMasterRequest
-            .newInstance(NetUtils.getHostname(), 1024, ""));
-
-    Assert.assertNotNull(responseRegister);
-    Assert.assertNotNull(responseRegister.getQueue());
-    Assert.assertNotNull(responseRegister.getApplicationACLs());
-    Assert.assertNotNull(responseRegister.getClientToAMTokenMasterKey());
-    Assert
-        .assertNotNull(responseRegister.getContainersFromPreviousAttempts());
-    Assert.assertNotNull(responseRegister.getSchedulerResourceTypes());
-    Assert.assertNotNull(responseRegister.getMaximumResourceCapability());
-
-    RMApp rmApp =
-        cluster.getResourceManager().getRMContext().getRMApps().get(appId);
-    Assert.assertEquals(RMAppState.RUNNING, rmApp.getState());
-
-    LOG.info("testDistributedSchedulingE2E - Allocate");
-
-    AllocateRequest request =
-        createAllocateRequest(rmClient.getNodeReports(NodeState.RUNNING));
-    List<ResourceRequest> askList = request.getAskList();
-    List<ResourceRequest> newAskList = new ArrayList<>(askList);
-
-    // Duplicate all ANY requests marking them as opportunistic
-    for (ResourceRequest rr : askList) {
-      if (ResourceRequest.ANY.equals(rr.getResourceName())) {
-        ResourceRequest newRR = ResourceRequest.newInstance(rr
-                .getPriority(), rr.getResourceName(),
-            rr.getCapability(), rr.getNumContainers(), rr.getRelaxLocality(),
-            rr.getNodeLabelExpression(),
-            ExecutionTypeRequest.newInstance(
-                ExecutionType.OPPORTUNISTIC, true));
-        newAskList.add(newRR);
-      }
-    }
-    request.setAskList(newAskList);
-
-    AllocateResponse allocResponse = client.allocate(request);
-    Assert.assertNotNull(allocResponse);
-
-    // Ensure that all the requests are satisfied immediately
-    Assert.assertEquals(2, allocResponse.getAllocatedContainers().size());
-
-    // Verify that the allocated containers are OPPORTUNISTIC
-    for (Container allocatedContainer : allocResponse
-        .getAllocatedContainers()) {
-      ContainerTokenIdentifier containerTokenIdentifier = BuilderUtils
-          .newContainerTokenIdentifier(
-              allocatedContainer.getContainerToken());
-      Assert.assertEquals(ExecutionType.OPPORTUNISTIC,
-          containerTokenIdentifier.getExecutionType());
-    }
-
-    request.setAskList(new ArrayList<ResourceRequest>());
-    request.setResponseId(request.getResponseId() + 1);
-
-    Thread.sleep(1000);
-
-    // RM should allocate GUARANTEED containers within 2 calls to allocate()
-    allocResponse = client.allocate(request);
-    Assert.assertNotNull(allocResponse);
-    Assert.assertEquals(2, allocResponse.getAllocatedContainers().size());
-
-    // Verify that the allocated containers are GUARANTEED
-    for (Container allocatedContainer : allocResponse
-        .getAllocatedContainers()) {
-      ContainerTokenIdentifier containerTokenIdentifier = BuilderUtils
-          .newContainerTokenIdentifier(
-              allocatedContainer.getContainerToken());
-      Assert.assertEquals(ExecutionType.GUARANTEED,
-          containerTokenIdentifier.getExecutionType());
-    }
-
-    LOG.info("testDistributedSchedulingE2E - Finish");
-  }
-
-  /**
-   * Validates if AMRMClient can be used with Distributed Scheduling turned on.
-   *
-   * @throws Exception
-   */
-  @Test(timeout = 120000)
-  @SuppressWarnings("unchecked")
-  public void testAMRMClient() throws Exception {
-    AMRMClientImpl<AMRMClient.ContainerRequest> amClient = null;
-    try {
-      Priority priority = Priority.newInstance(1);
-      Priority priority2 = Priority.newInstance(2);
-      Resource capability = Resource.newInstance(1024, 1);
-
-      List<NodeReport> nodeReports = rmClient.getNodeReports(NodeState.RUNNING);
-      String node = nodeReports.get(0).getNodeId().getHost();
-      String rack = nodeReports.get(0).getRackName();
-      String[] nodes = new String[]{node};
-      String[] racks = new String[]{rack};
-
-      // start am rm client
-      amClient = new AMRMClientImpl(client);
-      amClient.init(yarnConf);
-      amClient.start();
-      amClient.registerApplicationMaster(NetUtils.getHostname(), 1024, "");
-
-      assertEquals(0, amClient.ask.size());
-      assertEquals(0, amClient.release.size());
-
-      amClient.addContainerRequest(
-          new AMRMClient.ContainerRequest(capability, nodes, racks, priority));
-      amClient.addContainerRequest(
-          new AMRMClient.ContainerRequest(capability, nodes, racks, priority));
-      amClient.addContainerRequest(
-          new AMRMClient.ContainerRequest(capability, nodes, racks, priority));
-      amClient.addContainerRequest(
-          new AMRMClient.ContainerRequest(capability, nodes, racks, priority));
-      amClient.addContainerRequest(
-          new AMRMClient.ContainerRequest(capability, null, null, priority2,
-              0, true, null,
-              ExecutionTypeRequest.newInstance(
-                  ExecutionType.OPPORTUNISTIC, true)));
-      amClient.addContainerRequest(
-          new AMRMClient.ContainerRequest(capability, null, null, priority2,
-              0, true, null,
-              ExecutionTypeRequest.newInstance(
-                  ExecutionType.OPPORTUNISTIC, true)));
-
-      amClient.removeContainerRequest(
-          new AMRMClient.ContainerRequest(capability, nodes, racks, priority));
-      amClient.removeContainerRequest(
-          new AMRMClient.ContainerRequest(capability, nodes, racks, priority));
-      amClient.removeContainerRequest(
-          new AMRMClient.ContainerRequest(capability, null, null, priority2,
-              0, true, null,
-              ExecutionTypeRequest.newInstance(
-                  ExecutionType.OPPORTUNISTIC, true)));
-
-      RemoteRequestsTable<ContainerRequest> remoteRequestsTable =
-          amClient.getTable(0);
-      int containersRequestedNode = remoteRequestsTable.get(priority,
-          node, ExecutionType.GUARANTEED, capability).remoteRequest
-          .getNumContainers();
-      int containersRequestedRack = remoteRequestsTable.get(priority,
-          rack, ExecutionType.GUARANTEED, capability).remoteRequest
-          .getNumContainers();
-      int containersRequestedAny = remoteRequestsTable.get(priority,
-          ResourceRequest.ANY, ExecutionType.GUARANTEED, capability)
-          .remoteRequest.getNumContainers();
-      int oppContainersRequestedAny =
-          remoteRequestsTable.get(priority2, ResourceRequest.ANY,
-              ExecutionType.OPPORTUNISTIC, capability).remoteRequest
-              .getNumContainers();
-
-      assertEquals(2, containersRequestedNode);
-      assertEquals(2, containersRequestedRack);
-      assertEquals(2, containersRequestedAny);
-      assertEquals(1, oppContainersRequestedAny);
-
-      assertEquals(4, amClient.ask.size());
-      assertEquals(0, amClient.release.size());
-
-      // RM should allocate container within 2 calls to allocate()
-      int allocatedContainerCount = 0;
-      int iterationsLeft = 10;
-      Set<ContainerId> releases = new TreeSet<>();
-
-      amClient.getNMTokenCache().clearCache();
-      Assert.assertEquals(0,
-          amClient.getNMTokenCache().numberOfTokensInCache());
-      HashMap<String, Token> receivedNMTokens = new HashMap<>();
-
-      while (allocatedContainerCount <
-          (containersRequestedAny + oppContainersRequestedAny)
-          && iterationsLeft-- > 0) {
-        AllocateResponse allocResponse = amClient.allocate(0.1f);
-        assertEquals(0, amClient.ask.size());
-        assertEquals(0, amClient.release.size());
-
-        allocatedContainerCount += allocResponse.getAllocatedContainers()
-            .size();
-        for (Container container : allocResponse.getAllocatedContainers()) {
-          ContainerId rejectContainerId = container.getId();
-          releases.add(rejectContainerId);
-        }
-
-        for (NMToken token : allocResponse.getNMTokens()) {
-          String nodeID = token.getNodeId().toString();
-          receivedNMTokens.put(nodeID, token.getToken());
-        }
-
-        if (allocatedContainerCount < containersRequestedAny) {
-          // sleep to let NM's heartbeat to RM and trigger allocations
-          sleep(100);
-        }
-      }
-
-      assertEquals(allocatedContainerCount,
-          containersRequestedAny + oppContainersRequestedAny);
-      for (ContainerId rejectContainerId : releases) {
-        amClient.releaseAssignedContainer(rejectContainerId);
-      }
-      assertEquals(3, amClient.release.size());
-      assertEquals(0, amClient.ask.size());
-
-      // need to tell the AMRMClient that we dont need these resources anymore
-      amClient.removeContainerRequest(
-          new AMRMClient.ContainerRequest(capability, nodes, racks, priority));
-      amClient.removeContainerRequest(
-          new AMRMClient.ContainerRequest(capability, nodes, racks, priority));
-      amClient.removeContainerRequest(
-          new AMRMClient.ContainerRequest(capability, nodes, racks, priority2,
-              0, true, null,
-              ExecutionTypeRequest.newInstance(
-                  ExecutionType.OPPORTUNISTIC, true)));
-      assertEquals(4, amClient.ask.size());
-
-      // test RPC exception handling
-      amClient.addContainerRequest(new AMRMClient.ContainerRequest(capability,
-          nodes, racks, priority));
-      amClient.addContainerRequest(new AMRMClient.ContainerRequest(capability,
-          nodes, racks, priority));
-      amClient.addContainerRequest(
-          new AMRMClient.ContainerRequest(capability, nodes, racks, priority2,
-              0, true, null,
-              ExecutionTypeRequest.newInstance(
-                  ExecutionType.OPPORTUNISTIC, true)));
-
-      final AMRMClient amc = amClient;
-      ApplicationMasterProtocol realRM = amClient.rmClient;
-      try {
-        ApplicationMasterProtocol mockRM = mock(ApplicationMasterProtocol
-            .class);
-        final Resource _capability = capability;
-        final String[] _nodes = nodes;
-        final String[] _racks = racks;
-        final Priority _priority = priority;
-        final Priority _priority2 = priority2;
-
-        when(mockRM.allocate(any(AllocateRequest.class))).thenAnswer(
-            new Answer<AllocateResponse>() {
-              public AllocateResponse answer(InvocationOnMock invocation)
-                  throws Exception {
-                amc.removeContainerRequest(
-                    new AMRMClient.ContainerRequest(_capability, _nodes,
-                        _racks, _priority));
-                amc.removeContainerRequest(
-                    new AMRMClient.ContainerRequest(_capability, _nodes, _racks,
-                        _priority));
-                amc.removeContainerRequest(
-                    new AMRMClient.ContainerRequest(_capability, null, null,
-                        _priority2, 0, true, null,
-                        ExecutionTypeRequest.newInstance(
-                            ExecutionType.OPPORTUNISTIC, true)));
-                throw new Exception();
-              }
-            });
-        amClient.rmClient = mockRM;
-        amClient.allocate(0.1f);
-      } catch (Exception ioe) {
-      } finally {
-        amClient.rmClient = realRM;
-      }
-
-      assertEquals(3, amClient.release.size());
-      assertEquals(6, amClient.ask.size());
-
-      iterationsLeft = 3;
-      // do a few iterations to ensure RM is not going send new containers
-      while (iterationsLeft-- > 0) {
-        // inform RM of rejection
-        AllocateResponse allocResponse = amClient.allocate(0.1f);
-        // RM did not send new containers because AM does not need any
-        assertEquals(0, allocResponse.getAllocatedContainers().size());
-        if (allocResponse.getCompletedContainersStatuses().size() > 0) {
-          for (ContainerStatus cStatus : allocResponse
-              .getCompletedContainersStatuses()) {
-            if (releases.contains(cStatus.getContainerId())) {
-              assertEquals(cStatus.getState(), ContainerState.COMPLETE);
-              assertEquals(-100, cStatus.getExitStatus());
-              releases.remove(cStatus.getContainerId());
-            }
-          }
-        }
-        if (iterationsLeft > 0) {
-          // sleep to make sure NM's heartbeat
-          sleep(100);
-        }
-      }
-      assertEquals(0, amClient.ask.size());
-      assertEquals(0, amClient.release.size());
-
-      amClient.unregisterApplicationMaster(FinalApplicationStatus.SUCCEEDED,
-          null, null);
-
-    } finally {
-      if (amClient != null && amClient.getServiceState() == Service.STATE
-          .STARTED) {
-        amClient.stop();
-      }
-    }
-  }
-
-  /**
-   * Check if an AM can ask for opportunistic containers and get them.
-   * @throws Exception
-   */
-  @Test
-  public void testAMOpportunistic() throws Exception {
-    // Basic container to request
-    Resource capability = Resource.newInstance(1024, 1);
-    Priority priority = Priority.newInstance(1);
-
-    // Get the cluster topology
-    List<NodeReport> nodeReports = rmClient.getNodeReports(NodeState.RUNNING);
-    String node = nodeReports.get(0).getNodeId().getHost();
-    String rack = nodeReports.get(0).getRackName();
-    String[] nodes = new String[]{node};
-    String[] racks = new String[]{rack};
-
-    // Create an AM to request resources
-    AMRMClient<AMRMClient.ContainerRequest> amClient = null;
-    try {
-      amClient = new AMRMClientImpl<AMRMClient.ContainerRequest>(client);
-      amClient.init(yarnConf);
-      amClient.start();
-      amClient.registerApplicationMaster(NetUtils.getHostname(), 1024, "");
-
-      // AM requests an opportunistic container
-      ExecutionTypeRequest execTypeRequest =
-          ExecutionTypeRequest.newInstance(ExecutionType.OPPORTUNISTIC, true);
-      ContainerRequest containerRequest = new AMRMClient.ContainerRequest(
-          capability, nodes, racks, priority, 0, true, null, execTypeRequest);
-      amClient.addContainerRequest(containerRequest);
-
-      // Wait until the container is allocated
-      ContainerId opportunisticContainerId = null;
-      for (int i=0; i<10 && opportunisticContainerId == null; i++) {
-        AllocateResponse allocResponse = amClient.allocate(0.1f);
-        List<Container> allocatedContainers =
-            allocResponse.getAllocatedContainers();
-        for (Container allocatedContainer : allocatedContainers) {
-          // Check that this is the container we required
-          assertEquals(ExecutionType.OPPORTUNISTIC,
-              allocatedContainer.getExecutionType());
-          opportunisticContainerId = allocatedContainer.getId();
-        }
-        sleep(100);
-      }
-      assertNotNull(opportunisticContainerId);
-
-      // The RM sees the container as OPPORTUNISTIC
-      ResourceScheduler scheduler = cluster.getResourceManager()
-          .getResourceScheduler();
-      RMContainer rmContainer = scheduler.getRMContainer(
-          opportunisticContainerId);
-      assertEquals(ExecutionType.OPPORTUNISTIC,
-          rmContainer.getExecutionType());
-
-      // Release the opportunistic container
-      amClient.releaseAssignedContainer(opportunisticContainerId);
-      // Wait for the release container to appear
-      boolean released = false;
-      for (int i=0; i<10 && !released; i++) {
-        AllocateResponse allocResponse = amClient.allocate(0.1f);
-        List<ContainerStatus> completedContainers =
-            allocResponse.getCompletedContainersStatuses();
-        for (ContainerStatus completedContainer : completedContainers) {
-          ContainerId completedContainerId =
-              completedContainer.getContainerId();
-          assertEquals(completedContainerId, opportunisticContainerId);
-          released = true;
-        }
-        if (!released) {
-          sleep(100);
-        }
-      }
-      assertTrue(released);
-
-      // The RM shouldn't see the container anymore
-      rmContainer = scheduler.getRMContainer(opportunisticContainerId);
-      assertNull(rmContainer);
-
-      // Clean the AM
-      amClient.unregisterApplicationMaster(
-          FinalApplicationStatus.SUCCEEDED, null, null);
-    } finally {
-      if (amClient != null &&
-          amClient.getServiceState() == Service.STATE.STARTED) {
-        amClient.close();
-      }
-    }
-  }
-
-  private void sleep(int sleepTime) {
-    try {
-      Thread.sleep(sleepTime);
-    } catch (InterruptedException e) {
-      e.printStackTrace();
-    }
-  }
-}

http://git-wip-us.apache.org/repos/asf/hadoop/blob/cb91bb3c/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/test/java/org/apache/hadoop/yarn/client/api/impl/TestOpportunisticContainerAllocation.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/test/java/org/apache/hadoop/yarn/client/api/impl/TestOpportunisticContainerAllocation.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/test/java/org/apache/hadoop/yarn/client/api/impl/TestOpportunisticContainerAllocation.java
deleted file mode 100644
index 305d18b..0000000
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/test/java/org/apache/hadoop/yarn/client/api/impl/TestOpportunisticContainerAllocation.java
+++ /dev/null
@@ -1,781 +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
- * <p>
- * http://www.apache.org/licenses/LICENSE-2.0
- * <p>
- * 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.client.api.impl;
-
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.security.UserGroupInformation;
-import org.apache.hadoop.service.Service;
-import org.apache.hadoop.yarn.api.protocolrecords.AllocateResponse;
-import org.apache.hadoop.yarn.api.protocolrecords.SubmitApplicationRequest;
-import org.apache.hadoop.yarn.api.records.ApplicationAccessType;
-import org.apache.hadoop.yarn.api.records.ApplicationAttemptId;
-import org.apache.hadoop.yarn.api.records.ApplicationId;
-import org.apache.hadoop.yarn.api.records.ApplicationReport;
-import org.apache.hadoop.yarn.api.records.ApplicationSubmissionContext;
-import org.apache.hadoop.yarn.api.records.Container;
-import org.apache.hadoop.yarn.api.records.ContainerId;
-import org.apache.hadoop.yarn.api.records.ContainerLaunchContext;
-import org.apache.hadoop.yarn.api.records.ContainerState;
-import org.apache.hadoop.yarn.api.records.ContainerStatus;
-import org.apache.hadoop.yarn.api.records.ContainerUpdateType;
-import org.apache.hadoop.yarn.api.records.ExecutionType;
-import org.apache.hadoop.yarn.api.records.ExecutionTypeRequest;
-import org.apache.hadoop.yarn.api.records.FinalApplicationStatus;
-import org.apache.hadoop.yarn.api.records.LocalResource;
-import org.apache.hadoop.yarn.api.records.NMToken;
-import org.apache.hadoop.yarn.api.records.NodeReport;
-import org.apache.hadoop.yarn.api.records.NodeState;
-import org.apache.hadoop.yarn.api.records.Priority;
-import org.apache.hadoop.yarn.api.records.Resource;
-import org.apache.hadoop.yarn.api.records.ResourceRequest;
-import org.apache.hadoop.yarn.api.records.Token;
-import org.apache.hadoop.yarn.api.records.UpdateContainerRequest;
-import org.apache.hadoop.yarn.api.records.UpdatedContainer;
-import org.apache.hadoop.yarn.api.records.YarnApplicationState;
-import org.apache.hadoop.yarn.client.ClientRMProxy;
-import org.apache.hadoop.yarn.client.api.AMRMClient;
-import org.apache.hadoop.yarn.client.api.NMTokenCache;
-import org.apache.hadoop.yarn.client.api.YarnClient;
-import org.apache.hadoop.yarn.conf.YarnConfiguration;
-import org.apache.hadoop.yarn.exceptions.YarnException;
-import org.apache.hadoop.yarn.server.MiniYARNCluster;
-import org.apache.hadoop.yarn.server.resourcemanager.rmapp.attempt.RMAppAttempt;
-import org.apache.hadoop.yarn.server.resourcemanager.rmapp.attempt.RMAppAttemptState;
-
-import org.apache.hadoop.yarn.server.resourcemanager.scheduler
-    .AbstractYarnScheduler;
-import org.apache.hadoop.yarn.server.utils.BuilderUtils;
-import org.apache.hadoop.yarn.util.Records;
-import org.junit.After;
-import org.junit.AfterClass;
-import org.junit.Assert;
-import org.junit.Before;
-import org.junit.BeforeClass;
-import org.junit.Test;
-
-import java.io.IOException;
-import java.nio.ByteBuffer;
-import java.util.Arrays;
-import java.util.Collection;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
-import java.util.TreeSet;
-
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertNotNull;
-
-/**
- * Class that tests the allocation of OPPORTUNISTIC containers through the
- * centralized ResourceManager.
- */
-public class TestOpportunisticContainerAllocation {
-  private static Configuration conf = null;
-  private static MiniYARNCluster yarnCluster = null;
-  private static YarnClient yarnClient = null;
-  private static List<NodeReport> nodeReports = null;
-  private static int nodeCount = 3;
-
-  private static final int ROLLING_INTERVAL_SEC = 13;
-  private static final long AM_EXPIRE_MS = 4000;
-
-  private static Resource capability;
-  private static Priority priority;
-  private static Priority priority2;
-  private static Priority priority3;
-  private static Priority priority4;
-  private static String node;
-  private static String rack;
-  private static String[] nodes;
-  private static String[] racks;
-  private final static int DEFAULT_ITERATION = 3;
-
-  // Per test..
-  private ApplicationAttemptId attemptId = null;
-  private AMRMClientImpl<AMRMClient.ContainerRequest> amClient = null;
-  private long availMB;
-  private int availVCores;
-  private long allocMB;
-  private int allocVCores;
-
-  @BeforeClass
-  public static void setup() throws Exception {
-    // start minicluster
-    conf = new YarnConfiguration();
-    conf.setLong(
-        YarnConfiguration.RM_AMRM_TOKEN_MASTER_KEY_ROLLING_INTERVAL_SECS,
-        ROLLING_INTERVAL_SEC);
-    conf.setLong(YarnConfiguration.RM_AM_EXPIRY_INTERVAL_MS, AM_EXPIRE_MS);
-    conf.setInt(YarnConfiguration.RM_NM_HEARTBEAT_INTERVAL_MS, 1000);
-    // set the minimum allocation so that resource decrease can go under 1024
-    conf.setInt(YarnConfiguration.RM_SCHEDULER_MINIMUM_ALLOCATION_MB, 512);
-    conf.setBoolean(
-        YarnConfiguration.OPPORTUNISTIC_CONTAINER_ALLOCATION_ENABLED, true);
-    conf.setInt(
-        YarnConfiguration.NM_OPPORTUNISTIC_CONTAINERS_MAX_QUEUE_LENGTH, 10);
-    conf.setLong(YarnConfiguration.NM_LOG_RETAIN_SECONDS, 1);
-    yarnCluster =
-        new MiniYARNCluster(TestAMRMClient.class.getName(), nodeCount, 1, 1);
-    yarnCluster.init(conf);
-    yarnCluster.start();
-
-    // start rm client
-    yarnClient = YarnClient.createYarnClient();
-    yarnClient.init(conf);
-    yarnClient.start();
-
-    // get node info
-    nodeReports = yarnClient.getNodeReports(NodeState.RUNNING);
-
-    priority = Priority.newInstance(1);
-    priority2 = Priority.newInstance(2);
-    priority3 = Priority.newInstance(3);
-    priority4 = Priority.newInstance(4);
-    capability = Resource.newInstance(512, 1);
-
-    node = nodeReports.get(0).getNodeId().getHost();
-    rack = nodeReports.get(0).getRackName();
-    nodes = new String[]{node};
-    racks = new String[]{rack};
-  }
-
-  @Before
-  public void startApp() throws Exception {
-    // submit new app
-    ApplicationSubmissionContext appContext =
-        yarnClient.createApplication().getApplicationSubmissionContext();
-    ApplicationId appId = appContext.getApplicationId();
-    // set the application name
-    appContext.setApplicationName("Test");
-    // Set the priority for the application master
-    Priority pri = Records.newRecord(Priority.class);
-    pri.setPriority(0);
-    appContext.setPriority(pri);
-    // Set the queue to which this application is to be submitted in the RM
-    appContext.setQueue("default");
-    // Set up the container launch context for the application master
-    ContainerLaunchContext amContainer = BuilderUtils.newContainerLaunchContext(
-        Collections.<String, LocalResource>emptyMap(),
-        new HashMap<String, String>(), Arrays.asList("sleep", "100"),
-        new HashMap<String, ByteBuffer>(), null,
-        new HashMap<ApplicationAccessType, String>());
-    appContext.setAMContainerSpec(amContainer);
-    appContext.setResource(Resource.newInstance(1024, 1));
-    // Create the request to send to the applications manager
-    SubmitApplicationRequest appRequest =
-        Records.newRecord(SubmitApplicationRequest.class);
-    appRequest.setApplicationSubmissionContext(appContext);
-    // Submit the application to the applications manager
-    yarnClient.submitApplication(appContext);
-
-    // wait for app to start
-    RMAppAttempt appAttempt = null;
-    while (true) {
-      ApplicationReport appReport = yarnClient.getApplicationReport(appId);
-      if (appReport.getYarnApplicationState() ==
-          YarnApplicationState.ACCEPTED) {
-        attemptId = appReport.getCurrentApplicationAttemptId();
-        appAttempt = yarnCluster.getResourceManager().getRMContext().getRMApps()
-            .get(attemptId.getApplicationId()).getCurrentAppAttempt();
-        while (true) {
-          if (appAttempt.getAppAttemptState() == RMAppAttemptState.LAUNCHED) {
-            break;
-          }
-        }
-        break;
-      }
-    }
-    // Just dig into the ResourceManager and get the AMRMToken just for the sake
-    // of testing.
-    UserGroupInformation.setLoginUser(UserGroupInformation
-        .createRemoteUser(UserGroupInformation.getCurrentUser().getUserName()));
-
-    // emulate RM setup of AMRM token in credentials by adding the token
-    // *before* setting the token service
-    UserGroupInformation.getCurrentUser().addToken(appAttempt.getAMRMToken());
-    appAttempt.getAMRMToken()
-        .setService(ClientRMProxy.getAMRMTokenService(conf));
-
-    // start am rm client
-    amClient = (AMRMClientImpl<AMRMClient.ContainerRequest>)AMRMClient
-        .createAMRMClient();
-
-    //setting an instance NMTokenCache
-    amClient.setNMTokenCache(new NMTokenCache());
-    //asserting we are not using the singleton instance cache
-    Assert.assertNotSame(NMTokenCache.getSingleton(),
-        amClient.getNMTokenCache());
-
-    amClient.init(conf);
-    amClient.start();
-
-    amClient.registerApplicationMaster("Host", 10000, "");
-  }
-
-  @After
-  public void cancelApp() throws YarnException, IOException {
-    try {
-      amClient
-          .unregisterApplicationMaster(FinalApplicationStatus.SUCCEEDED, null,
-              null);
-    } finally {
-      if (amClient != null &&
-          amClient.getServiceState() == Service.STATE.STARTED) {
-        amClient.stop();
-      }
-    }
-    yarnClient.killApplication(attemptId.getApplicationId());
-    attemptId = null;
-  }
-
-  @AfterClass
-  public static void tearDown() {
-    if (yarnClient != null &&
-        yarnClient.getServiceState() == Service.STATE.STARTED) {
-      yarnClient.stop();
-    }
-    if (yarnCluster != null &&
-        yarnCluster.getServiceState() == Service.STATE.STARTED) {
-      yarnCluster.stop();
-    }
-  }
-
-  @Test(timeout = 60000)
-  public void testPromotionFromAcquired() throws YarnException, IOException {
-    // setup container request
-    assertEquals(0, amClient.ask.size());
-    assertEquals(0, amClient.release.size());
-
-    amClient.addContainerRequest(
-        new AMRMClient.ContainerRequest(capability, null, null, priority2, 0,
-            true, null,
-            ExecutionTypeRequest.newInstance(
-                ExecutionType.OPPORTUNISTIC, true)));
-
-    int oppContainersRequestedAny =
-        amClient.getTable(0).get(priority2, ResourceRequest.ANY,
-            ExecutionType.OPPORTUNISTIC, capability).remoteRequest
-            .getNumContainers();
-
-    assertEquals(1, oppContainersRequestedAny);
-
-    assertEquals(1, amClient.ask.size());
-    assertEquals(0, amClient.release.size());
-
-    // RM should allocate container within 2 calls to allocate()
-    int allocatedContainerCount = 0;
-    Map<ContainerId, Container> allocatedOpportContainers = new HashMap<>();
-    int iterationsLeft = 50;
-
-    amClient.getNMTokenCache().clearCache();
-    Assert.assertEquals(0,
-        amClient.getNMTokenCache().numberOfTokensInCache());
-    HashMap<String, Token> receivedNMTokens = new HashMap<>();
-
-    updateMetrics("Before Opp Allocation");
-
-    while (allocatedContainerCount < oppContainersRequestedAny
-        && iterationsLeft-- > 0) {
-      AllocateResponse allocResponse = amClient.allocate(0.1f);
-      assertEquals(0, amClient.ask.size());
-      assertEquals(0, amClient.release.size());
-
-      allocatedContainerCount +=
-          allocResponse.getAllocatedContainers().size();
-      for (Container container : allocResponse.getAllocatedContainers()) {
-        if (container.getExecutionType() == ExecutionType.OPPORTUNISTIC) {
-          allocatedOpportContainers.put(container.getId(), container);
-          removeCR(container);
-        }
-      }
-
-      for (NMToken token : allocResponse.getNMTokens()) {
-        String nodeID = token.getNodeId().toString();
-        receivedNMTokens.put(nodeID, token.getToken());
-      }
-
-      if (allocatedContainerCount < oppContainersRequestedAny) {
-        // sleep to let NM's heartbeat to RM and trigger allocations
-        sleep(100);
-      }
-    }
-
-    assertEquals(oppContainersRequestedAny, allocatedContainerCount);
-    assertEquals(oppContainersRequestedAny, allocatedOpportContainers.size());
-
-    updateMetrics("After Opp Allocation / Before Promotion");
-
-    try {
-      Container c = allocatedOpportContainers.values().iterator().next();
-      amClient.requestContainerUpdate(
-          c, UpdateContainerRequest.newInstance(c.getVersion(),
-              c.getId(), ContainerUpdateType.PROMOTE_EXECUTION_TYPE,
-              null, ExecutionType.OPPORTUNISTIC));
-      Assert.fail("Should throw Exception..");
-    } catch (IllegalArgumentException e) {
-      System.out.println("## " + e.getMessage());
-      Assert.assertTrue(e.getMessage().contains(
-          "target should be GUARANTEED and original should be OPPORTUNISTIC"));
-    }
-
-    Container c = allocatedOpportContainers.values().iterator().next();
-    amClient.requestContainerUpdate(
-        c, UpdateContainerRequest.newInstance(c.getVersion(),
-            c.getId(), ContainerUpdateType.PROMOTE_EXECUTION_TYPE,
-            null, ExecutionType.GUARANTEED));
-    iterationsLeft = 120;
-    Map<ContainerId, UpdatedContainer> updatedContainers = new HashMap<>();
-    // do a few iterations to ensure RM is not going to send new containers
-    while (iterationsLeft-- > 0 && updatedContainers.isEmpty()) {
-      // inform RM of rejection
-      AllocateResponse allocResponse = amClient.allocate(0.1f);
-      // RM did not send new containers because AM does not need any
-      if (allocResponse.getUpdatedContainers() != null) {
-        for (UpdatedContainer updatedContainer : allocResponse
-            .getUpdatedContainers()) {
-          System.out.println("Got update..");
-          updatedContainers.put(updatedContainer.getContainer().getId(),
-              updatedContainer);
-        }
-      }
-      if (iterationsLeft > 0) {
-        // sleep to make sure NM's heartbeat
-        sleep(100);
-      }
-    }
-
-    updateMetrics("After Promotion");
-
-    assertEquals(1, updatedContainers.size());
-    for (ContainerId cId : allocatedOpportContainers.keySet()) {
-      Container orig = allocatedOpportContainers.get(cId);
-      UpdatedContainer updatedContainer = updatedContainers.get(cId);
-      assertNotNull(updatedContainer);
-      assertEquals(ExecutionType.GUARANTEED,
-          updatedContainer.getContainer().getExecutionType());
-      assertEquals(orig.getResource(),
-          updatedContainer.getContainer().getResource());
-      assertEquals(orig.getNodeId(),
-          updatedContainer.getContainer().getNodeId());
-      assertEquals(orig.getVersion() + 1,
-          updatedContainer.getContainer().getVersion());
-    }
-    assertEquals(0, amClient.ask.size());
-    assertEquals(0, amClient.release.size());
-    amClient.ask.clear();
-  }
-
-  @Test(timeout = 60000)
-  public void testDemotionFromAcquired() throws YarnException, IOException {
-    // setup container request
-    assertEquals(0, amClient.ask.size());
-    assertEquals(0, amClient.release.size());
-
-    amClient.addContainerRequest(
-        new AMRMClient.ContainerRequest(capability, null, null, priority3));
-
-    int guarContainersRequestedAny = amClient.getTable(0).get(priority3,
-        ResourceRequest.ANY, ExecutionType.GUARANTEED, capability)
-        .remoteRequest.getNumContainers();
-
-    assertEquals(1, guarContainersRequestedAny);
-
-    assertEquals(1, amClient.ask.size());
-    assertEquals(0, amClient.release.size());
-
-    // RM should allocate container within 2 calls to allocate()
-    int allocatedContainerCount = 0;
-    Map<ContainerId, Container> allocatedGuarContainers = new HashMap<>();
-    int iterationsLeft = 50;
-
-    amClient.getNMTokenCache().clearCache();
-    Assert.assertEquals(0,
-        amClient.getNMTokenCache().numberOfTokensInCache());
-    HashMap<String, Token> receivedNMTokens = new HashMap<>();
-
-    updateMetrics("Before Guar Allocation");
-
-    while (allocatedContainerCount < guarContainersRequestedAny
-        && iterationsLeft-- > 0) {
-      AllocateResponse allocResponse = amClient.allocate(0.1f);
-      assertEquals(0, amClient.ask.size());
-      assertEquals(0, amClient.release.size());
-
-      allocatedContainerCount +=
-          allocResponse.getAllocatedContainers().size();
-      for (Container container : allocResponse.getAllocatedContainers()) {
-        if (container.getExecutionType() == ExecutionType.GUARANTEED) {
-          allocatedGuarContainers.put(container.getId(), container);
-          removeCR(container);
-        }
-      }
-
-      for (NMToken token : allocResponse.getNMTokens()) {
-        String nodeID = token.getNodeId().toString();
-        receivedNMTokens.put(nodeID, token.getToken());
-      }
-
-      if (allocatedContainerCount < guarContainersRequestedAny) {
-        // sleep to let NM's heartbeat to RM and trigger allocations
-        sleep(100);
-      }
-    }
-
-    assertEquals(guarContainersRequestedAny, allocatedContainerCount);
-    assertEquals(guarContainersRequestedAny, allocatedGuarContainers.size());
-
-    updateMetrics("After Guar Allocation / Before Demotion");
-
-    try {
-      Container c = allocatedGuarContainers.values().iterator().next();
-      amClient.requestContainerUpdate(
-          c, UpdateContainerRequest.newInstance(c.getVersion(),
-              c.getId(), ContainerUpdateType.DEMOTE_EXECUTION_TYPE,
-              null, ExecutionType.GUARANTEED));
-      Assert.fail("Should throw Exception..");
-    } catch (IllegalArgumentException e) {
-      System.out.println("## " + e.getMessage());
-      Assert.assertTrue(e.getMessage().contains(
-          "target should be OPPORTUNISTIC and original should be GUARANTEED"));
-    }
-
-    Container c = allocatedGuarContainers.values().iterator().next();
-    amClient.requestContainerUpdate(
-        c, UpdateContainerRequest.newInstance(c.getVersion(),
-            c.getId(), ContainerUpdateType.DEMOTE_EXECUTION_TYPE,
-            null, ExecutionType.OPPORTUNISTIC));
-    iterationsLeft = 120;
-    Map<ContainerId, UpdatedContainer> updatedContainers = new HashMap<>();
-    // do a few iterations to ensure RM is not going to send new containers
-    while (iterationsLeft-- > 0 && updatedContainers.isEmpty()) {
-      // inform RM of rejection
-      AllocateResponse allocResponse = amClient.allocate(0.1f);
-      // RM did not send new containers because AM does not need any
-      if (allocResponse.getUpdatedContainers() != null) {
-        for (UpdatedContainer updatedContainer : allocResponse
-            .getUpdatedContainers()) {
-          System.out.println("Got update..");
-          updatedContainers.put(updatedContainer.getContainer().getId(),
-              updatedContainer);
-        }
-      }
-      if (iterationsLeft > 0) {
-        // sleep to make sure NM's heartbeat
-        sleep(100);
-      }
-    }
-
-    updateMetrics("After Demotion");
-
-    assertEquals(1, updatedContainers.size());
-    for (ContainerId cId : allocatedGuarContainers.keySet()) {
-      Container orig = allocatedGuarContainers.get(cId);
-      UpdatedContainer updatedContainer = updatedContainers.get(cId);
-      assertNotNull(updatedContainer);
-      assertEquals(ExecutionType.OPPORTUNISTIC,
-          updatedContainer.getContainer().getExecutionType());
-      assertEquals(orig.getResource(),
-          updatedContainer.getContainer().getResource());
-      assertEquals(orig.getNodeId(),
-          updatedContainer.getContainer().getNodeId());
-      assertEquals(orig.getVersion() + 1,
-          updatedContainer.getContainer().getVersion());
-    }
-    assertEquals(0, amClient.ask.size());
-    assertEquals(0, amClient.release.size());
-    amClient.ask.clear();
-  }
-
-  @Test(timeout = 60000)
-  public void testMixedAllocationAndRelease() throws YarnException,
-      IOException {
-    // setup container request
-    assertEquals(0, amClient.ask.size());
-    assertEquals(0, amClient.release.size());
-
-    amClient.addContainerRequest(
-        new AMRMClient.ContainerRequest(capability, nodes, racks, priority));
-    amClient.addContainerRequest(
-        new AMRMClient.ContainerRequest(capability, nodes, racks, priority));
-    amClient.addContainerRequest(
-        new AMRMClient.ContainerRequest(capability, nodes, racks, priority));
-    amClient.addContainerRequest(
-        new AMRMClient.ContainerRequest(capability, nodes, racks, priority));
-    amClient.addContainerRequest(
-        new AMRMClient.ContainerRequest(capability, null, null, priority2, 0,
-            true, null,
-            ExecutionTypeRequest.newInstance(
-                ExecutionType.OPPORTUNISTIC, true)));
-    amClient.addContainerRequest(
-        new AMRMClient.ContainerRequest(capability, null, null, priority2, 0,
-            true, null,
-            ExecutionTypeRequest.newInstance(
-                ExecutionType.OPPORTUNISTIC, true)));
-
-    int containersRequestedNode = amClient.getTable(0).get(priority,
-        node, ExecutionType.GUARANTEED, capability).remoteRequest
-        .getNumContainers();
-    int containersRequestedRack = amClient.getTable(0).get(priority,
-        rack, ExecutionType.GUARANTEED, capability).remoteRequest
-        .getNumContainers();
-    int containersRequestedAny = amClient.getTable(0).get(priority,
-        ResourceRequest.ANY, ExecutionType.GUARANTEED, capability)
-        .remoteRequest.getNumContainers();
-    int oppContainersRequestedAny =
-        amClient.getTable(0).get(priority2, ResourceRequest.ANY,
-            ExecutionType.OPPORTUNISTIC, capability).remoteRequest
-            .getNumContainers();
-
-    assertEquals(4, containersRequestedNode);
-    assertEquals(4, containersRequestedRack);
-    assertEquals(4, containersRequestedAny);
-    assertEquals(2, oppContainersRequestedAny);
-
-    assertEquals(4, amClient.ask.size());
-    assertEquals(0, amClient.release.size());
-
-    amClient.removeContainerRequest(
-        new AMRMClient.ContainerRequest(capability, nodes, racks, priority));
-    amClient.removeContainerRequest(
-        new AMRMClient.ContainerRequest(capability, nodes, racks, priority));
-    amClient.removeContainerRequest(
-        new AMRMClient.ContainerRequest(capability, null, null, priority2, 0,
-            true, null,
-            ExecutionTypeRequest.newInstance(
-                ExecutionType.OPPORTUNISTIC, true)));
-
-    containersRequestedNode = amClient.getTable(0).get(priority,
-        node, ExecutionType.GUARANTEED, capability).remoteRequest
-        .getNumContainers();
-    containersRequestedRack = amClient.getTable(0).get(priority,
-        rack, ExecutionType.GUARANTEED, capability).remoteRequest
-        .getNumContainers();
-    containersRequestedAny = amClient.getTable(0).get(priority,
-        ResourceRequest.ANY, ExecutionType.GUARANTEED, capability)
-        .remoteRequest.getNumContainers();
-    oppContainersRequestedAny =
-        amClient.getTable(0).get(priority2, ResourceRequest.ANY,
-            ExecutionType.OPPORTUNISTIC, capability).remoteRequest
-            .getNumContainers();
-
-    assertEquals(2, containersRequestedNode);
-    assertEquals(2, containersRequestedRack);
-    assertEquals(2, containersRequestedAny);
-    assertEquals(1, oppContainersRequestedAny);
-
-    assertEquals(4, amClient.ask.size());
-    assertEquals(0, amClient.release.size());
-
-    // RM should allocate container within 2 calls to allocate()
-    int allocatedContainerCount = 0;
-    int allocatedOpportContainerCount = 0;
-    int iterationsLeft = 50;
-    Set<ContainerId> releases = new TreeSet<>();
-
-    amClient.getNMTokenCache().clearCache();
-    Assert.assertEquals(0,
-        amClient.getNMTokenCache().numberOfTokensInCache());
-    HashMap<String, Token> receivedNMTokens = new HashMap<>();
-
-    while (allocatedContainerCount <
-        containersRequestedAny + oppContainersRequestedAny
-        && iterationsLeft-- > 0) {
-      AllocateResponse allocResponse = amClient.allocate(0.1f);
-      assertEquals(0, amClient.ask.size());
-      assertEquals(0, amClient.release.size());
-
-      allocatedContainerCount +=
-          allocResponse.getAllocatedContainers().size();
-      for (Container container : allocResponse.getAllocatedContainers()) {
-        if (container.getExecutionType() == ExecutionType.OPPORTUNISTIC) {
-          allocatedOpportContainerCount++;
-        }
-        ContainerId rejectContainerId = container.getId();
-        releases.add(rejectContainerId);
-      }
-
-      for (NMToken token : allocResponse.getNMTokens()) {
-        String nodeID = token.getNodeId().toString();
-        receivedNMTokens.put(nodeID, token.getToken());
-      }
-
-      if (allocatedContainerCount < containersRequestedAny) {
-        // sleep to let NM's heartbeat to RM and trigger allocations
-        sleep(100);
-      }
-    }
-
-    assertEquals(containersRequestedAny + oppContainersRequestedAny,
-        allocatedContainerCount);
-    assertEquals(oppContainersRequestedAny, allocatedOpportContainerCount);
-    for (ContainerId rejectContainerId : releases) {
-      amClient.releaseAssignedContainer(rejectContainerId);
-    }
-    assertEquals(3, amClient.release.size());
-    assertEquals(0, amClient.ask.size());
-
-    // need to tell the AMRMClient that we don't need these resources anymore
-    amClient.removeContainerRequest(
-        new AMRMClient.ContainerRequest(capability, nodes, racks, priority));
-    amClient.removeContainerRequest(
-        new AMRMClient.ContainerRequest(capability, nodes, racks, priority));
-    amClient.removeContainerRequest(
-        new AMRMClient.ContainerRequest(capability, nodes, racks, priority2, 0,
-            true, null,
-            ExecutionTypeRequest.newInstance(
-                ExecutionType.OPPORTUNISTIC, true)));
-    assertEquals(4, amClient.ask.size());
-
-    iterationsLeft = 3;
-    // do a few iterations to ensure RM is not going to send new containers
-    while (iterationsLeft-- > 0) {
-      // inform RM of rejection
-      AllocateResponse allocResponse = amClient.allocate(0.1f);
-      // RM did not send new containers because AM does not need any
-      assertEquals(0, allocResponse.getAllocatedContainers().size());
-      if (allocResponse.getCompletedContainersStatuses().size() > 0) {
-        for (ContainerStatus cStatus : allocResponse
-            .getCompletedContainersStatuses()) {
-          if (releases.contains(cStatus.getContainerId())) {
-            assertEquals(cStatus.getState(), ContainerState.COMPLETE);
-            assertEquals(-100, cStatus.getExitStatus());
-            releases.remove(cStatus.getContainerId());
-          }
-        }
-      }
-      if (iterationsLeft > 0) {
-        // sleep to make sure NM's heartbeat
-        sleep(100);
-      }
-    }
-    assertEquals(0, amClient.ask.size());
-    assertEquals(0, amClient.release.size());
-  }
-
-  /**
-   * Tests allocation with requests comprising only opportunistic containers.
-   */
-  @Test(timeout = 60000)
-  public void testOpportunisticAllocation() throws YarnException, IOException {
-    // setup container request
-    assertEquals(0, amClient.ask.size());
-    assertEquals(0, amClient.release.size());
-
-    amClient.addContainerRequest(
-        new AMRMClient.ContainerRequest(capability, null, null, priority3, 0,
-            true, null,
-            ExecutionTypeRequest.newInstance(
-                ExecutionType.OPPORTUNISTIC, true)));
-    amClient.addContainerRequest(
-        new AMRMClient.ContainerRequest(capability, null, null, priority3, 0,
-            true, null,
-            ExecutionTypeRequest.newInstance(
-                ExecutionType.OPPORTUNISTIC, true)));
-
-    int oppContainersRequestedAny =
-        amClient.getTable(0).get(priority3, ResourceRequest.ANY,
-            ExecutionType.OPPORTUNISTIC, capability).remoteRequest
-            .getNumContainers();
-
-    assertEquals(2, oppContainersRequestedAny);
-
-    assertEquals(1, amClient.ask.size());
-    assertEquals(0, amClient.release.size());
-
-    // RM should allocate container within 2 calls to allocate()
-    int allocatedContainerCount = 0;
-    int iterationsLeft = 10;
-    Set<ContainerId> releases = new TreeSet<>();
-
-    amClient.getNMTokenCache().clearCache();
-    Assert.assertEquals(0,
-        amClient.getNMTokenCache().numberOfTokensInCache());
-    HashMap<String, Token> receivedNMTokens = new HashMap<>();
-
-    while (allocatedContainerCount < oppContainersRequestedAny
-        && iterationsLeft-- > 0) {
-      AllocateResponse allocResponse = amClient.allocate(0.1f);
-      assertEquals(0, amClient.ask.size());
-      assertEquals(0, amClient.release.size());
-
-      for (Container container : allocResponse.getAllocatedContainers()) {
-        allocatedContainerCount++;
-        ContainerId rejectContainerId = container.getId();
-        releases.add(rejectContainerId);
-      }
-
-      for (NMToken token : allocResponse.getNMTokens()) {
-        String nodeID = token.getNodeId().toString();
-        receivedNMTokens.put(nodeID, token.getToken());
-      }
-
-      if (allocatedContainerCount < oppContainersRequestedAny) {
-        // sleep to let NM's heartbeat to RM and trigger allocations
-        sleep(100);
-      }
-    }
-
-    assertEquals(oppContainersRequestedAny, allocatedContainerCount);
-    assertEquals(1, receivedNMTokens.values().size());
-  }
-
-  private void removeCR(Container container) {
-    List<? extends Collection<AMRMClient.ContainerRequest>>
-        matchingRequests = amClient.getMatchingRequests(container
-            .getPriority(),
-        ResourceRequest.ANY, ExecutionType.OPPORTUNISTIC,
-        container.getResource());
-    Set<AMRMClient.ContainerRequest> toRemove = new HashSet<>();
-    for (Collection<AMRMClient.ContainerRequest> rc : matchingRequests) {
-      for (AMRMClient.ContainerRequest cr : rc) {
-        toRemove.add(cr);
-      }
-    }
-    for (AMRMClient.ContainerRequest cr : toRemove) {
-      amClient.removeContainerRequest(cr);
-    }
-  }
-
-  private void updateMetrics(String msg) {
-    AbstractYarnScheduler scheduler =
-        (AbstractYarnScheduler)yarnCluster.getResourceManager()
-            .getResourceScheduler();
-    availMB = scheduler.getRootQueueMetrics().getAvailableMB();
-    availVCores = scheduler.getRootQueueMetrics().getAvailableVirtualCores();
-    allocMB = scheduler.getRootQueueMetrics().getAllocatedMB();
-    allocVCores = scheduler.getRootQueueMetrics().getAllocatedVirtualCores();
-    System.out.println("## METRICS (" + msg + ")==>");
-    System.out.println(" : availMB=" + availMB + ", " +
-        "availVCores=" +availVCores + ", " +
-        "allocMB=" + allocMB + ", " +
-        "allocVCores=" + allocVCores + ", ");
-    System.out.println("<== ##");
-  }
-
-  private void sleep(int sleepTime) {
-    try {
-      Thread.sleep(sleepTime);
-    } catch (InterruptedException e) {
-      e.printStackTrace();
-    }
-  }
-}


---------------------------------------------------------------------
To unsubscribe, e-mail: common-commits-unsubscribe@hadoop.apache.org
For additional commands, e-mail: common-commits-help@hadoop.apache.org


Mime
View raw message