MAPREDUCE-6809. Create ContainerRequestor interface and refactor RMContainerRequestor to use it. Contributed by Devaraj K.

hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapreduce/v2/app/MRAppMaster.java

@@ -111,7 +111,6 @@ import org.apache.hadoop.mapreduce.v2.app.rm.ContainerAllocator;
 import org.apache.hadoop.mapreduce.v2.app.rm.ContainerAllocatorEvent;
 import org.apache.hadoop.mapreduce.v2.app.rm.RMCommunicator;
 import org.apache.hadoop.mapreduce.v2.app.rm.RMContainerAllocator;
-import org.apache.hadoop.mapreduce.v2.app.rm.RMContainerRequestor;
 import org.apache.hadoop.mapreduce.v2.app.rm.RMHeartbeatHandler;
 import org.apache.hadoop.mapreduce.v2.app.rm.preemption.AMPreemptionPolicy;
 import org.apache.hadoop.mapreduce.v2.app.rm.preemption.NoopAMPreemptionPolicy;
@@ -1156,7 +1155,7 @@ public class MRAppMaster extends CompositeService {
 
     @Override
     public Set<String> getBlacklistedNodes() {
-      return ((RMContainerRequestor) containerAllocator).getBlacklistedNodes();
+      return ((RMContainerAllocator) containerAllocator).getBlacklistedNodes();
     }
     
     @Override

hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapreduce/v2/app/rm/ContainerRequestor.java

@@ -0,0 +1,50 @@
+/**
+ * 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.mapreduce.v2.app.rm;
+
+import java.io.IOException;
+
+import org.apache.hadoop.mapreduce.v2.app.rm.RMContainerRequestor.ContainerRequest;
+import org.apache.hadoop.yarn.api.protocolrecords.AllocateResponse;
+import org.apache.hadoop.yarn.api.records.ContainerId;
+import org.apache.hadoop.yarn.api.records.Resource;
+import org.apache.hadoop.yarn.exceptions.YarnException;
+import org.apache.hadoop.yarn.exceptions.YarnRuntimeException;
+
+/**
+ * Interface for ContainerReqestor.
+ */
+public interface ContainerRequestor {
+
+  AllocateResponse makeRemoteRequest()
+      throws YarnRuntimeException, YarnException, IOException;
+
+  void addContainerReq(ContainerRequest request);
+
+  void decContainerReq(ContainerRequest request);
+
+  void release(ContainerId containerId);
+
+  boolean isNodeBlacklisted(String hostname);
+
+  Resource getAvailableResources();
+
+  void containerFailedOnHost(String hostName);
+
+  ContainerRequest filterRequest(ContainerRequest orig);
+}

hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapreduce/v2/app/rm/RMContainerAllocator.java

@@ -58,6 +58,7 @@ import org.apache.hadoop.mapreduce.v2.app.job.event.TaskAttemptDiagnosticsUpdate
 import org.apache.hadoop.mapreduce.v2.app.job.event.TaskAttemptEvent;
 import org.apache.hadoop.mapreduce.v2.app.job.event.TaskAttemptEventType;
 import org.apache.hadoop.mapreduce.v2.app.job.event.TaskAttemptKillEvent;
+import org.apache.hadoop.mapreduce.v2.app.rm.RMContainerRequestor.ContainerRequest;
 import org.apache.hadoop.mapreduce.v2.app.rm.preemption.AMPreemptionPolicy;
 import org.apache.hadoop.security.UserGroupInformation;
 import org.apache.hadoop.util.StringInterner;
@@ -94,7 +95,7 @@ import org.slf4j.LoggerFactory;
 /**
  * Allocates the container from the ResourceManager scheduler.
  */
-public class RMContainerAllocator extends RMContainerRequestor
+public class RMContainerAllocator extends RMCommunicator
     implements ContainerAllocator {
 
   static final Logger LOG = LoggerFactory.getLogger(RMContainerAllocator.class);
@@ -114,6 +115,9 @@ public class RMContainerAllocator extends RMContainerRequestor
   private Thread eventHandlingThread;
   private final AtomicBoolean stopped;
 
+  @VisibleForTesting
+  protected RMContainerRequestor containerRequestor;
+
   static {
     PRIORITY_FAST_FAIL_MAP = RecordFactoryProvider.getRecordFactory(null).newRecordInstance(Priority.class);
     PRIORITY_FAST_FAIL_MAP.setPriority(5);
@@ -207,6 +211,7 @@ public class RMContainerAllocator extends RMContainerRequestor
     return new AssignedRequests();
   }
 
+  @SuppressWarnings("unchecked")
   @Override
   protected void serviceInit(Configuration conf) throws Exception {
     super.serviceInit(conf);
@@ -242,6 +247,8 @@ public class RMContainerAllocator extends RMContainerRequestor
             MRJobConfig.DEFAULT_MR_NUM_OPPORTUNISTIC_MAPS_PERCENT));
     LOG.info(this.scheduledRequests.getNumOpportunisticMapsPercent() +
         "% of the mappers will be scheduled using OPPORTUNISTIC containers");
+    containerRequestor = new RMContainerRequestor(this);
+    containerRequestor.init(conf);
   }
 
   @Override
@@ -398,8 +405,8 @@ public class RMContainerAllocator extends RMContainerRequestor
           removed = true;
           assignedRequests.remove(aId);
           containersReleased++;
-          pendingRelease.add(containerId);
-          release(containerId);
+          containerRequestor.pendingRelease.add(containerId);
+          containerRequestor.release(containerId);
         }
       }
       if (!removed) {
@@ -411,7 +418,7 @@ public class RMContainerAllocator extends RMContainerRequestor
         event.getType() == ContainerAllocator.EventType.CONTAINER_FAILED) {
       ContainerFailedEvent fEv = (ContainerFailedEvent) event;
       String host = getHost(fEv.getContMgrAddress());
-      containerFailedOnHost(host);
+      containerRequestor.containerFailedOnHost(host);
       // propagate failures to preemption policy to discard checkpoints for
       // failed tasks
       preemptionPolicy.handleFailedContainer(event.getAttemptID());
@@ -569,13 +576,14 @@ public class RMContainerAllocator extends RMContainerRequestor
     // there are enough resources for a mapper to run. This is calculated by
     // excluding scheduled reducers from headroom and comparing it against
     // resources required to run one mapper.
-    Resource scheduledReducesResource = Resources.multiply(
-         reduceResourceRequest, scheduledRequests.reduces.size());
-    Resource availableResourceForMap =
-         Resources.subtract(getAvailableResources(), scheduledReducesResource);
-    if (ResourceCalculatorUtils.computeAvailableContainers(availableResourceForMap,
-        mapResourceRequest, getSchedulerResourceTypes()) > 0) {
-       // Enough room to run a mapper
+    Resource scheduledReducesResource = Resources
+        .multiply(reduceResourceRequest, scheduledRequests.reduces.size());
+    Resource availableResourceForMap = Resources.subtract(
+        containerRequestor.getAvailableResources(), scheduledReducesResource);
+    if (ResourceCalculatorUtils.computeAvailableContainers(
+        availableResourceForMap, mapResourceRequest,
+        getSchedulerResourceTypes()) > 0) {
+      // Enough room to run a mapper
       return false;
     }
 
@@ -651,7 +659,7 @@ public class RMContainerAllocator extends RMContainerRequestor
     }
     
     // get available resources for this job
-    Resource headRoom = getAvailableResources();
+    Resource headRoom = containerRequestor.getAvailableResources();
 
     LOG.info("Recalculating schedule, headroom=" + headRoom);
     
@@ -782,7 +790,8 @@ public class RMContainerAllocator extends RMContainerRequestor
     applyConcurrentTaskLimits();
 
     // will be null the first time
-    Resource headRoom = Resources.clone(getAvailableResources());
+    Resource headRoom = Resources
+        .clone(containerRequestor.getAvailableResources());
     AllocateResponse response;
     /*
      * If contact with RM is lost, the AM will wait MR_AM_TO_RM_WAIT_INTERVAL_MS
@@ -790,7 +799,7 @@ public class RMContainerAllocator extends RMContainerRequestor
      * to contact the RM.
      */
     try {
-      response = makeRemoteRequest();
+      response = containerRequestor.makeRemoteRequest();
       // Reset retry count if no exception occurred.
       retrystartTime = System.currentTimeMillis();
     } catch (ApplicationAttemptNotFoundException e ) {
@@ -805,9 +814,9 @@ public class RMContainerAllocator extends RMContainerRequestor
       LOG.info("ApplicationMaster is out of sync with ResourceManager,"
           + " hence resync and send outstanding requests.");
       // RM may have restarted, re-register with RM.
-      lastResponseID = 0;
+      containerRequestor.lastResponseID = 0;
       register();
-      addOutstandingRequestOnResync();
+      containerRequestor.addOutstandingRequestOnResync();
       return null;
     } catch (InvalidLabelResourceRequestException e) {
       // If Invalid label exception is received means the requested label doesnt
@@ -833,7 +842,7 @@ public class RMContainerAllocator extends RMContainerRequestor
       // continue to attempt to contact the RM.
       throw e;
     }
-    Resource newHeadRoom = getAvailableResources();
+    Resource newHeadRoom = containerRequestor.getAvailableResources();
     List<Container> newContainers = response.getAllocatedContainers();
     // Setting NMTokens
     if (response.getNMTokens() != null) {
@@ -874,7 +883,7 @@ public class RMContainerAllocator extends RMContainerRequestor
     }
 
     //Called on each allocation. Will know about newly blacklisted/added hosts.
-    computeIgnoreBlacklisting();
+    containerRequestor.computeIgnoreBlacklisting();
 
     handleUpdatedNodes(response);
     handleJobPriorityChange(response);
@@ -900,7 +909,7 @@ public class RMContainerAllocator extends RMContainerRequestor
       LOG.error("Container complete event for unknown container "
           + container.getContainerId());
     } else {
-      pendingRelease.remove(container.getContainerId());
+      containerRequestor.pendingRelease.remove(container.getContainerId());
       assignedRequests.remove(attemptID);
 
       // Send the diagnostics
@@ -927,11 +936,12 @@ public class RMContainerAllocator extends RMContainerRequestor
       int normalMapRequestLimit = Math.min(
           maxRequestedMaps - failedMapRequestLimit,
           numScheduledMaps - numScheduledFailMaps);
-      setRequestLimit(PRIORITY_FAST_FAIL_MAP, mapResourceRequest,
-          failedMapRequestLimit);
-      setRequestLimit(PRIORITY_MAP, mapResourceRequest, normalMapRequestLimit);
-      setRequestLimit(PRIORITY_OPPORTUNISTIC_MAP, mapResourceRequest,
+      containerRequestor.setRequestLimit(PRIORITY_FAST_FAIL_MAP,
+          mapResourceRequest, failedMapRequestLimit);
+      containerRequestor.setRequestLimit(PRIORITY_MAP, mapResourceRequest,
           normalMapRequestLimit);
+      containerRequestor.setRequestLimit(PRIORITY_OPPORTUNISTIC_MAP,
+          mapResourceRequest, normalMapRequestLimit);
     }
 
     int numScheduledReduces = scheduledRequests.reduces.size();
@@ -941,7 +951,7 @@ public class RMContainerAllocator extends RMContainerRequestor
           maxRunningReduces - assignedRequests.reduces.size());
       int reduceRequestLimit = Math.min(maxRequestedReduces,
           numScheduledReduces);
-      setRequestLimit(PRIORITY_REDUCE, reduceResourceRequest,
+      containerRequestor.setRequestLimit(PRIORITY_REDUCE, reduceResourceRequest,
           reduceRequestLimit);
     }
   }
@@ -1036,7 +1046,7 @@ public class RMContainerAllocator extends RMContainerRequestor
 
   @Private
   public Resource getResourceLimit() {
-    Resource headRoom = getAvailableResources();
+    Resource headRoom = containerRequestor.getAvailableResources();
     Resource assignedMapResource =
         Resources.multiply(mapResourceRequest, assignedRequests.maps.size());
     Resource assignedReduceResource =
@@ -1087,7 +1097,7 @@ public class RMContainerAllocator extends RMContainerRequestor
       if (req == null) {
         return false;
       } else {
-        decContainerReq(req);
+        containerRequestor.decContainerReq(req);
         return true;
       }
     }
@@ -1097,7 +1107,7 @@ public class RMContainerAllocator extends RMContainerRequestor
       if (it.hasNext()) {
         Entry<TaskAttemptId, ContainerRequest> entry = it.next();
         it.remove();
-        decContainerReq(entry.getValue());
+        containerRequestor.decContainerReq(entry.getValue());
         return entry.getValue();
       }
       return null;
@@ -1114,14 +1124,15 @@ public class RMContainerAllocator extends RMContainerRequestor
         LOG.info("Added "+event.getAttemptID()+" to list of failed maps");
         // If its an earlier Failed attempt, do not retry as OPPORTUNISTIC
         maps.put(event.getAttemptID(), request);
-        addContainerReq(request);
+        containerRequestor.addContainerReq(request);
       } else {
         if (mapsMod100 < numOpportunisticMapsPercent) {
           request =
               new ContainerRequest(event, PRIORITY_OPPORTUNISTIC_MAP,
                   mapNodeLabelExpression);
           maps.put(event.getAttemptID(), request);
-          addOpportunisticResourceRequest(request.priority, request.capability);
+          containerRequestor.addOpportunisticResourceRequest(request.priority,
+              request.capability);
         } else {
           request =
               new ContainerRequest(event, PRIORITY_MAP, mapNodeLabelExpression);
@@ -1148,7 +1159,7 @@ public class RMContainerAllocator extends RMContainerRequestor
             }
           }
           maps.put(event.getAttemptID(), request);
-          addContainerReq(request);
+          containerRequestor.addContainerReq(request);
         }
         mapsMod100++;
         mapsMod100 %= 100;
@@ -1158,7 +1169,7 @@ public class RMContainerAllocator extends RMContainerRequestor
     
     void addReduce(ContainerRequest req) {
       reduces.put(req.attemptID, req);
-      addContainerReq(req);
+      containerRequestor.addContainerReq(req);
     }
     
     // this method will change the list of allocatedContainers.
@@ -1223,7 +1234,7 @@ public class RMContainerAllocator extends RMContainerRequestor
         // do not assign if allocated container is on a  
         // blacklisted host
         String allocatedHost = allocated.getNodeId().getHost();
-        if (isNodeBlacklisted(allocatedHost)) {
+        if (containerRequestor.isNodeBlacklisted(allocatedHost)) {
           // we need to request for a new container 
           // and release the current one
           LOG.info("Got allocated container on a blacklisted "
@@ -1237,9 +1248,9 @@ public class RMContainerAllocator extends RMContainerRequestor
           if (toBeReplacedReq != null) {
             LOG.info("Placing a new container request for task attempt " 
                 + toBeReplacedReq.attemptID);
-            ContainerRequest newReq = 
-                getFilteredContainerRequest(toBeReplacedReq);
-            decContainerReq(toBeReplacedReq);
+            ContainerRequest newReq = containerRequestor
+                .filterRequest(toBeReplacedReq);
+            containerRequestor.decContainerReq(toBeReplacedReq);
             if (toBeReplacedReq.attemptID.getTaskId().getTaskType() ==
                 TaskType.MAP) {
               maps.put(newReq.attemptID, newReq);
@@ -1247,7 +1258,7 @@ public class RMContainerAllocator extends RMContainerRequestor
             else {
               reduces.put(newReq.attemptID, newReq);
             }
-            addContainerReq(newReq);
+            containerRequestor.addContainerReq(newReq);
           }
           else {
             LOG.info("Could not map allocated container to a valid request."
@@ -1276,7 +1287,7 @@ public class RMContainerAllocator extends RMContainerRequestor
     private void containerAssigned(Container allocated, 
                                     ContainerRequest assigned) {
       // Update resource requests
-      decContainerReq(assigned);
+      containerRequestor.decContainerReq(assigned);
 
       // send the container-assigned event to task attempt
       eventHandler.handle(new TaskAttemptContainerAssignedEvent(
@@ -1293,8 +1304,8 @@ public class RMContainerAllocator extends RMContainerRequestor
     
     private void containerNotAssigned(Container allocated) {
       containersReleased++;
-      pendingRelease.add(allocated.getId());
-      release(allocated.getId());      
+      containerRequestor.pendingRelease.add(allocated.getId());
+      containerRequestor.release(allocated.getId());
     }
     
     private ContainerRequest assignWithoutLocality(Container allocated) {
@@ -1659,4 +1670,18 @@ public class RMContainerAllocator extends RMContainerRequestor
 
   }
 
+  public Set<String> getBlacklistedNodes() {
+    return containerRequestor.getBlacklistedNodes();
+  }
+
+  public RMContainerRequestor getContainerRequestor() {
+    return containerRequestor;
+  }
+
+  public void resetContainerForReuse(ContainerId containerId) {
+    TaskAttemptId attemptId = assignedRequests.get(containerId);
+    if (attemptId != null) {
+      assignedRequests.remove(attemptId);
+    }
+  }
 }

hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapreduce/v2/app/rm/RMContainerRequestor.java

@@ -34,10 +34,10 @@ import org.apache.hadoop.classification.InterfaceAudience.Private;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.mapreduce.MRJobConfig;
 import org.apache.hadoop.mapreduce.v2.api.records.TaskAttemptId;
-import org.apache.hadoop.mapreduce.v2.app.AppContext;
-import org.apache.hadoop.mapreduce.v2.app.client.ClientService;
+import org.apache.hadoop.service.AbstractService;
 import org.apache.hadoop.yarn.api.protocolrecords.AllocateRequest;
 import org.apache.hadoop.yarn.api.protocolrecords.AllocateResponse;
+import org.apache.hadoop.yarn.api.records.ApplicationId;
 import org.apache.hadoop.yarn.api.records.ContainerId;
 import org.apache.hadoop.yarn.api.records.ExecutionTypeRequest;
 import org.apache.hadoop.yarn.api.records.ExecutionType;
@@ -60,7 +60,8 @@ import org.slf4j.LoggerFactory;
 /**
  * Keeps the data structures to send container requests to RM.
  */
-public abstract class RMContainerRequestor extends RMCommunicator {
+public class RMContainerRequestor extends AbstractService
+    implements ContainerRequestor {
   
   private static final Logger LOG =
       LoggerFactory.getLogger(RMContainerRequestor.class);
@@ -111,9 +112,13 @@ public abstract class RMContainerRequestor extends RMCommunicator {
       .newSetFromMap(new ConcurrentHashMap<String, Boolean>());
   private final Set<String> blacklistRemovals = Collections
       .newSetFromMap(new ConcurrentHashMap<String, Boolean>());
+  private final ApplicationId applicationId;
+  private final RMCommunicator rmCommunicator;
 
-  public RMContainerRequestor(ClientService clientService, AppContext context) {
-    super(clientService, context);
+  public RMContainerRequestor(RMCommunicator rmCommunicator) {
+    super(RMContainerRequestor.class.getName());
+    this.rmCommunicator = rmCommunicator;
+    applicationId = rmCommunicator.applicationId;
   }
 
   @Private
@@ -194,17 +199,19 @@ public abstract class RMContainerRequestor extends RMCommunicator {
     LOG.info("blacklistDisablePercent is " + blacklistDisablePercent);
   }
 
-  protected AllocateResponse makeRemoteRequest() throws YarnException,
-      IOException {
+  @Override
+  public AllocateResponse makeRemoteRequest()
+      throws YarnException, IOException {
     applyRequestLimits();
     ResourceBlacklistRequest blacklistRequest =
         ResourceBlacklistRequest.newInstance(new ArrayList<String>(blacklistAdditions),
             new ArrayList<String>(blacklistRemovals));
-    AllocateRequest allocateRequest =
-        AllocateRequest.newInstance(lastResponseID,
-          super.getApplicationProgress(), new ArrayList<ResourceRequest>(ask),
-          new ArrayList<ContainerId>(release), blacklistRequest);
-    AllocateResponse allocateResponse = scheduler.allocate(allocateRequest);
+    AllocateRequest allocateRequest = AllocateRequest.newInstance(
+        lastResponseID, rmCommunicator.getApplicationProgress(),
+        new ArrayList<ResourceRequest>(ask),
+        new ArrayList<ContainerId>(release), blacklistRequest);
+    AllocateResponse allocateResponse = rmCommunicator.scheduler
+        .allocate(allocateRequest);
     lastResponseID = allocateResponse.getResponseId();
     availableResources = allocateResponse.getAvailableResources();
     lastClusterNmCount = clusterNmCount;
@@ -324,7 +331,8 @@ public abstract class RMContainerRequestor extends RMCommunicator {
     }
   }
   
-  protected void containerFailedOnHost(String hostName) {
+  @Override
+  public void containerFailedOnHost(String hostName) {
     if (!nodeBlacklistingEnabled) {
       return;
     }
@@ -389,11 +397,13 @@ public abstract class RMContainerRequestor extends RMCommunicator {
     }
   }
 
-  protected Resource getAvailableResources() {
+  @Override
+  public Resource getAvailableResources() {
     return availableResources == null ? Resources.none() : availableResources;
   }
 
-  protected void addContainerReq(ContainerRequest req) {
+  @Override
+  public void addContainerReq(ContainerRequest req) {
     // Create resource requests
     for (String host : req.hosts) {
       // Data-local
@@ -414,7 +424,8 @@ public abstract class RMContainerRequestor extends RMCommunicator {
         req.nodeLabelExpression);
   }
 
-  protected void decContainerReq(ContainerRequest req) {
+  @Override
+  public void decContainerReq(ContainerRequest req) {
     // Update resource requests
     for (String hostName : req.hosts) {
       decResourceRequest(req.priority, hostName, req.capability);
@@ -540,18 +551,21 @@ public abstract class RMContainerRequestor extends RMCommunicator {
     ask.add(remoteRequest);    
   }
 
-  protected void release(ContainerId containerId) {
+  @Override
+  public void release(ContainerId containerId) {
     release.add(containerId);
   }
   
-  protected boolean isNodeBlacklisted(String hostname) {
+  @Override
+  public boolean isNodeBlacklisted(String hostname) {
     if (!nodeBlacklistingEnabled || ignoreBlacklisting.get()) {
       return false;
     }
     return blacklistedNodes.contains(hostname);
   }
-  
-  protected ContainerRequest getFilteredContainerRequest(ContainerRequest orig) {
+
+  @Override
+  public ContainerRequest filterRequest(ContainerRequest orig) {
     ArrayList<String> newHosts = new ArrayList<String>();
     for (String host : orig.hosts) {
       if (!isNodeBlacklisted(host)) {

hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/test/java/org/apache/hadoop/mapreduce/v2/app/rm/TestRMContainerAllocator.java

@@ -27,6 +27,7 @@ import static org.mockito.ArgumentMatchers.anyInt;
 import static org.mockito.ArgumentMatchers.isA;
 import static org.mockito.Mockito.doCallRealMethod;
 import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.doThrow;
 import static org.mockito.Mockito.inOrder;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.never;
@@ -674,7 +675,7 @@ public class TestRMContainerAllocator {
             Resource.newInstance(1024, 1),
             locations, false, true));
     allocator.scheduleAllReduces();
-    allocator.makeRemoteRequest();
+    allocator.containerRequestor.makeRemoteRequest();
     nm.nodeHeartbeat(true);
     rm.drainEvents();
     allocator.sendRequest(createRequest(jobId, 1,
@@ -2141,24 +2142,18 @@ public class TestRMContainerAllocator {
     public void updateSchedulerProxy(MyResourceManager rm) {
       scheduler = rm.getApplicationMasterService();
     }
-
-    @Override
-    protected AllocateResponse makeRemoteRequest() throws IOException,
-        YarnException {
-      allocateResponse = super.makeRemoteRequest();
-      return allocateResponse;
-    }
   }
 
   private static class MyContainerAllocator2 extends MyContainerAllocator {
     public MyContainerAllocator2(MyResourceManager rm, Configuration conf,
-      ApplicationAttemptId appAttemptId, Job job) {
+        ApplicationAttemptId appAttemptId, Job job)
+        throws YarnException, IOException {
       super(rm, conf, appAttemptId, job);
-    }
-    @Override
-    protected AllocateResponse makeRemoteRequest() throws IOException,
-      YarnException {
-      throw new IOException("for testing");
+      containerRequestor = mock(RMContainerRequestor.class);
+      doThrow(new IOException("for testing")).when(containerRequestor)
+          .makeRemoteRequest();
+      doReturn(Resource.newInstance(2048, 1)).when(containerRequestor)
+          .getAvailableResources();
     }
   }
 
@@ -2196,6 +2191,7 @@ public class TestRMContainerAllocator {
         any(Resource.class), anyInt(), anyFloat(), anyFloat());
     doReturn(EnumSet.of(SchedulerResourceTypes.MEMORY)).when(allocator)
       .getSchedulerResourceTypes();
+    allocator.containerRequestor = mock(RMContainerRequestor.class);
 
     // Test slow-start
     allocator.scheduleReduces(
@@ -2972,11 +2968,6 @@ public class TestRMContainerAllocator {
             return mockScheduler;
           }
 
-          @Override
-          protected void setRequestLimit(Priority priority,
-              Resource capability, int limit) {
-            Assert.fail("setRequestLimit() should not be invoked");
-          }
         };
 
     // create some map requests
@@ -3257,8 +3248,8 @@ public class TestRMContainerAllocator {
     Assert.assertEquals(1, allocator.getScheduledRequests().maps.size());
     Assert.assertEquals(0, allocator.getAssignedRequests().maps.size());
 
-    Assert.assertEquals(6, allocator.getAsk().size());
-    for (ResourceRequest req : allocator.getAsk()) {
+    Assert.assertEquals(6, allocator.containerRequestor.getAsk().size());
+    for (ResourceRequest req : allocator.containerRequestor.getAsk()) {
       boolean isReduce =
           req.getPriority().equals(RMContainerAllocator.PRIORITY_REDUCE);
       if (isReduce) {
@@ -3283,8 +3274,8 @@ public class TestRMContainerAllocator {
     // indicate ramping down of reduces to scheduler.
     Assert.assertEquals(0, allocator.getScheduledRequests().reduces.size());
     Assert.assertEquals(2, allocator.getNumOfPendingReduces());
-    Assert.assertEquals(3, allocator.getAsk().size());
-    for (ResourceRequest req : allocator.getAsk()) {
+    Assert.assertEquals(3, allocator.containerRequestor.getAsk().size());
+    for (ResourceRequest req : allocator.containerRequestor.getAsk()) {
       Assert.assertEquals(
           RMContainerAllocator.PRIORITY_REDUCE, req.getPriority());
       Assert.assertTrue(req.getResourceName().equals("*") ||
@@ -3311,6 +3302,7 @@ public class TestRMContainerAllocator {
     RMContainerAllocator containerAllocator =
         new RMContainerAllocatorForFinishedContainer(null, context,
             mock(AMPreemptionPolicy.class));
+    containerAllocator.init(new Configuration());
 
     ContainerStatus finishedContainer = ContainerStatus.newInstance(
         mock(ContainerId.class), ContainerState.COMPLETE, "", 0);
@@ -3427,8 +3419,8 @@ public class TestRMContainerAllocator {
     Assert.assertEquals(1, allocator.getScheduledRequests().maps.size());
     Assert.assertEquals(0, allocator.getAssignedRequests().maps.size());
 
-    Assert.assertEquals(6, allocator.getAsk().size());
-    for (ResourceRequest req : allocator.getAsk()) {
+    Assert.assertEquals(6, allocator.containerRequestor.getAsk().size());
+    for (ResourceRequest req : allocator.containerRequestor.getAsk()) {
       boolean isReduce =
           req.getPriority().equals(RMContainerAllocator.PRIORITY_REDUCE);
       if (isReduce) {
@@ -3456,8 +3448,8 @@ public class TestRMContainerAllocator {
     // indicate ramping down of reduces to scheduler.
     Assert.assertEquals(0, allocator.getScheduledRequests().reduces.size());
     Assert.assertEquals(2, allocator.getNumOfPendingReduces());
-    Assert.assertEquals(3, allocator.getAsk().size());
-    for (ResourceRequest req : allocator.getAsk()) {
+    Assert.assertEquals(3, allocator.containerRequestor.getAsk().size());
+    for (ResourceRequest req : allocator.containerRequestor.getAsk()) {
       Assert.assertEquals(
           RMContainerAllocator.PRIORITY_REDUCE, req.getPriority());
       Assert.assertTrue(req.getResourceName().equals("*") ||