1) Fixing yet another PB related race condition.

2) Also making release API to take container-id instead of container.

git-svn-id: https://svn.apache.org/repos/asf/hadoop/common/branches/MR-279@1153456 13f79535-47bb-0310-9956-ffa450edef68

mapreduce/mr-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapreduce/v2/app/rm/RMCommunicator.java

@@ -51,7 +51,7 @@ import org.apache.hadoop.yarn.api.protocolrecords.RegisterApplicationMasterRespo
 import org.apache.hadoop.yarn.api.records.AMResponse;
 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.ContainerId;
 import org.apache.hadoop.yarn.api.records.Resource;
 import org.apache.hadoop.yarn.api.records.ResourceRequest;
 import org.apache.hadoop.yarn.conf.YarnConfiguration;
@@ -269,7 +269,7 @@ public class RMCommunicator extends AbstractService  {
     allocateRequest.setApplicationAttemptId(applicationAttemptId);
     allocateRequest.setResponseId(lastResponseID);
     allocateRequest.addAllAsks(new ArrayList<ResourceRequest>());
-    allocateRequest.addAllReleases(new ArrayList<Container>());
+    allocateRequest.addAllReleases(new ArrayList<ContainerId>());
     AllocateResponse allocateResponse = scheduler.allocate(allocateRequest);
     AMResponse response = allocateResponse.getAMResponse();
   }

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

@@ -228,12 +228,12 @@ public class RMContainerAllocator extends RMContainerRequestor
       
       boolean removed = scheduledRequests.remove(aId);
       if (!removed) {
-        Container container = assignedRequests.get(aId);
-        if (container != null) {
+        ContainerId containerId = assignedRequests.get(aId);
+        if (containerId != null) {
           removed = true;
           assignedRequests.remove(aId);
           containersReleased++;
-          release(container);
+          release(containerId);
         }
       }
       if (!removed) {
@@ -565,16 +565,18 @@ public class RMContainerAllocator extends RMContainerRequestor
           eventHandler.handle(new TaskAttemptContainerAssignedEvent(
               assigned.attemptID, allocated));
 
-          assignedRequests.add(allocated, assigned.attemptID);
+          assignedRequests.add(allocated.getId(), assigned.attemptID);
           
           LOG.info("Assigned container (" + allocated + ") " +
               " to task " + assigned.attemptID +
               " on node " + allocated.getNodeId().toString());
         } else {
           //not assigned to any request, release the container
-          LOG.info("Releasing unassigned container " + allocated);
+          LOG.info("Releasing unassigned and invalid container " + allocated
+              + ". RM has gone crazy, someone go look!"
+              + " Hey RM, if you are so rich, go donate to non-profits!");
           containersReleased++;
-          release(allocated);
+          release(allocated.getId());
         }
       }
     }
@@ -705,21 +707,21 @@ public class RMContainerAllocator extends RMContainerRequestor
   private class AssignedRequests {
     private final Map<ContainerId, TaskAttemptId> containerToAttemptMap =
       new HashMap<ContainerId, TaskAttemptId>();
-    private final LinkedHashMap<TaskAttemptId, Container> maps = 
-      new LinkedHashMap<TaskAttemptId, Container>();
-    private final LinkedHashMap<TaskAttemptId, Container> reduces = 
-      new LinkedHashMap<TaskAttemptId, Container>();
+    private final LinkedHashMap<TaskAttemptId, ContainerId> maps = 
+      new LinkedHashMap<TaskAttemptId, ContainerId>();
+    private final LinkedHashMap<TaskAttemptId, ContainerId> reduces = 
+      new LinkedHashMap<TaskAttemptId, ContainerId>();
     private final Set<TaskAttemptId> preemptionWaitingReduces = 
       new HashSet<TaskAttemptId>();
     
-    void add(Container container, TaskAttemptId tId) {
-      LOG.info("Assigned container " + container.getNodeId().toString()
+    void add(ContainerId containerId, TaskAttemptId tId) {
+      LOG.info("Assigned container " + containerId.toString()
           + " to " + tId);
-      containerToAttemptMap.put(container.getId(), tId);
+      containerToAttemptMap.put(containerId, tId);
       if (tId.getTaskId().getTaskType().equals(TaskType.MAP)) {
-        maps.put(tId, container);
+        maps.put(tId, containerId);
       } else {
-        reduces.put(tId, container);
+        reduces.put(tId, containerId);
       }
     }
 
@@ -745,12 +747,12 @@ public class RMContainerAllocator extends RMContainerRequestor
     }
     
     boolean remove(TaskAttemptId tId) {
-      Container container = null;
+      ContainerId containerId = null;
       if (tId.getTaskId().getTaskType().equals(TaskType.MAP)) {
-        container = maps.remove(tId);
+        containerId = maps.remove(tId);
       } else {
-        container = reduces.remove(tId);
-        if (container != null) {
+        containerId = reduces.remove(tId);
+        if (containerId != null) {
           boolean preempted = preemptionWaitingReduces.remove(tId);
           if (preempted) {
             LOG.info("Reduce preemption successful " + tId);
@@ -758,8 +760,8 @@ public class RMContainerAllocator extends RMContainerRequestor
         }
       }
       
-      if (container != null) {
-        containerToAttemptMap.remove(container.getId());
+      if (containerId != null) {
+        containerToAttemptMap.remove(containerId);
         return true;
       }
       return false;
@@ -769,7 +771,7 @@ public class RMContainerAllocator extends RMContainerRequestor
       return containerToAttemptMap.get(cId);
     }
 
-    Container get(TaskAttemptId tId) {
+    ContainerId get(TaskAttemptId tId) {
       if (tId.getTaskId().getTaskType().equals(TaskType.MAP)) {
         return maps.get(tId);
       } else {

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

@@ -21,7 +21,6 @@ package org.apache.hadoop.mapreduce.v2.app.rm;
 import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.HashSet;
-import java.util.List;
 import java.util.Map;
 import java.util.Set;
 import java.util.TreeMap;
@@ -71,7 +70,7 @@ public abstract class RMContainerRequestor extends RMCommunicator {
       new TreeMap<Priority, Map<String, Map<Resource, ResourceRequest>>>();
 
   private final Set<ResourceRequest> ask = new TreeSet<ResourceRequest>();
-  private final Set<Container> release = new TreeSet<Container>(); 
+  private final Set<ContainerId> release = new TreeSet<ContainerId>(); 
 
   private boolean nodeBlacklistingEnabled;
   private int maxTaskFailuresPerNode;
@@ -118,7 +117,7 @@ public abstract class RMContainerRequestor extends RMCommunicator {
     allocateRequest.setApplicationAttemptId(applicationAttemptId);
     allocateRequest.setResponseId(lastResponseID);
     allocateRequest.addAllAsks(new ArrayList<ResourceRequest>(ask));
-    allocateRequest.addAllReleases(new ArrayList<Container>(release));
+    allocateRequest.addAllReleases(new ArrayList<ContainerId>(release));
     AllocateResponse allocateResponse = scheduler.allocate(allocateRequest);
     AMResponse response = allocateResponse.getAMResponse();
     lastResponseID = response.getResponseId();
@@ -268,8 +267,8 @@ public abstract class RMContainerRequestor extends RMCommunicator {
              + remoteRequest.getNumContainers() + " #asks=" + ask.size());
   }
 
-  protected void release(Container container) {
-    release.add(container);
+  protected void release(ContainerId containerId) {
+    release.add(containerId);
   }
   
 }

mapreduce/yarn/yarn-api/src/main/java/org/apache/hadoop/yarn/api/protocolrecords/AllocateRequest.java

@@ -3,7 +3,7 @@ package org.apache.hadoop.yarn.api.protocolrecords;
 import java.util.List;
 
 import org.apache.hadoop.yarn.api.records.ApplicationAttemptId;
-import org.apache.hadoop.yarn.api.records.Container;
+import org.apache.hadoop.yarn.api.records.ContainerId;
 import org.apache.hadoop.yarn.api.records.ResourceRequest;
 
 public interface AllocateRequest {
@@ -21,8 +21,8 @@ public interface AllocateRequest {
   ResourceRequest getAsk(int index);
   int getAskCount();
   
-  List<Container> getReleaseList();
-  Container getRelease(int index);
+  List<ContainerId> getReleaseList();
+  ContainerId getRelease(int index);
   int getReleaseCount();
 
   void addAllAsks(List<ResourceRequest> resourceRequest);
@@ -30,8 +30,8 @@ public interface AllocateRequest {
   void removeAsk(int index);
   void clearAsks();
   
-  void addAllReleases(List<Container> releaseContainers);
-  void addRelease(Container container);
+  void addAllReleases(List<ContainerId> releaseContainers);
+  void addRelease(ContainerId container);
   void removeRelease(int index);
   void clearReleases();
 }

mapreduce/yarn/yarn-api/src/main/java/org/apache/hadoop/yarn/api/protocolrecords/impl/pb/AllocateRequestPBImpl.java

@@ -8,13 +8,14 @@ import java.util.List;
 import org.apache.hadoop.yarn.api.protocolrecords.AllocateRequest;
 import org.apache.hadoop.yarn.api.records.ApplicationAttemptId;
 import org.apache.hadoop.yarn.api.records.Container;
+import org.apache.hadoop.yarn.api.records.ContainerId;
 import org.apache.hadoop.yarn.api.records.ProtoBase;
 import org.apache.hadoop.yarn.api.records.ResourceRequest;
 import org.apache.hadoop.yarn.api.records.impl.pb.ApplicationAttemptIdPBImpl;
-import org.apache.hadoop.yarn.api.records.impl.pb.ContainerPBImpl;
+import org.apache.hadoop.yarn.api.records.impl.pb.ContainerIdPBImpl;
 import org.apache.hadoop.yarn.api.records.impl.pb.ResourceRequestPBImpl;
 import org.apache.hadoop.yarn.proto.YarnProtos.ApplicationAttemptIdProto;
-import org.apache.hadoop.yarn.proto.YarnProtos.ContainerProto;
+import org.apache.hadoop.yarn.proto.YarnProtos.ContainerIdProto;
 import org.apache.hadoop.yarn.proto.YarnProtos.ResourceRequestProto;
 import org.apache.hadoop.yarn.proto.YarnServiceProtos.AllocateRequestProto;
 import org.apache.hadoop.yarn.proto.YarnServiceProtos.AllocateRequestProtoOrBuilder;
@@ -28,7 +29,7 @@ public class AllocateRequestPBImpl extends ProtoBase<AllocateRequestProto> imple
 
   private ApplicationAttemptId applicationAttemptID = null;
   private List<ResourceRequest> ask = null;
-  private List<Container> release = null;
+  private List<ContainerId> release = null;
   
   
   public AllocateRequestPBImpl() {
@@ -205,12 +206,12 @@ public class AllocateRequestPBImpl extends ProtoBase<AllocateRequestProto> imple
     this.ask.clear();
   }
   @Override
-  public List<Container> getReleaseList() {
+  public List<ContainerId> getReleaseList() {
     initReleases();
     return this.release;
   }
   @Override
-  public Container getRelease(int index) {
+  public ContainerId getRelease(int index) {
     initReleases();
     return this.release.get(index);
   }
@@ -225,16 +226,16 @@ public class AllocateRequestPBImpl extends ProtoBase<AllocateRequestProto> imple
       return;
     }
     AllocateRequestProtoOrBuilder p = viaProto ? proto : builder;
-    List<ContainerProto> list = p.getReleaseList();
-    this.release = new ArrayList<Container>();
+    List<ContainerIdProto> list = p.getReleaseList();
+    this.release = new ArrayList<ContainerId>();
 
-    for (ContainerProto c : list) {
+    for (ContainerIdProto c : list) {
       this.release.add(convertFromProtoFormat(c));
     }
   }
   
   @Override
-  public void addAllReleases(final List<Container> release) {
+  public void addAllReleases(final List<ContainerId> release) {
     if (release == null)
       return;
     initReleases();
@@ -246,12 +247,12 @@ public class AllocateRequestPBImpl extends ProtoBase<AllocateRequestProto> imple
     builder.clearRelease();
     if (release == null)
       return;
-    Iterable<ContainerProto> iterable = new Iterable<ContainerProto>() {
+    Iterable<ContainerIdProto> iterable = new Iterable<ContainerIdProto>() {
       @Override
-      public Iterator<ContainerProto> iterator() {
-        return new Iterator<ContainerProto>() {
+      public Iterator<ContainerIdProto> iterator() {
+        return new Iterator<ContainerIdProto>() {
 
-          Iterator<Container> iter = release.iterator();
+          Iterator<ContainerId> iter = release.iterator();
 
           @Override
           public boolean hasNext() {
@@ -259,7 +260,7 @@ public class AllocateRequestPBImpl extends ProtoBase<AllocateRequestProto> imple
           }
 
           @Override
-          public ContainerProto next() {
+          public ContainerIdProto next() {
             return convertToProtoFormat(iter.next());
           }
 
@@ -275,7 +276,7 @@ public class AllocateRequestPBImpl extends ProtoBase<AllocateRequestProto> imple
     builder.addAllRelease(iterable);
   }
   @Override
-  public void addRelease(Container release) {
+  public void addRelease(ContainerId release) {
     initReleases();
     this.release.add(release);
   }
@@ -306,11 +307,11 @@ public class AllocateRequestPBImpl extends ProtoBase<AllocateRequestProto> imple
     return ((ResourceRequestPBImpl)t).getProto();
   }
 
-  private ContainerPBImpl convertFromProtoFormat(ContainerProto p) {
-    return new ContainerPBImpl(p);
+  private ContainerIdPBImpl convertFromProtoFormat(ContainerIdProto p) {
+    return new ContainerIdPBImpl(p);
   }
 
-  private ContainerProto convertToProtoFormat(Container t) {
-    return ((ContainerPBImpl)t).getProto();
+  private ContainerIdProto convertToProtoFormat(ContainerId t) {
+    return ((ContainerIdPBImpl)t).getProto();
   }
 }  

mapreduce/yarn/yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/impl/pb/ContainerTokenPBImpl.java

@@ -11,9 +11,9 @@ import org.apache.hadoop.yarn.proto.YarnProtos.ContainerTokenProtoOrBuilder;
 
     
 public class ContainerTokenPBImpl extends ProtoBase<ContainerTokenProto> implements ContainerToken {
-  ContainerTokenProto proto = ContainerTokenProto.getDefaultInstance();
-  ContainerTokenProto.Builder builder = null;
-  boolean viaProto = false;
+  private ContainerTokenProto proto = ContainerTokenProto.getDefaultInstance();
+  private ContainerTokenProto.Builder builder = null;
+  private boolean viaProto = false;
   
   private ByteBuffer identifier;
   private ByteBuffer password;
@@ -28,14 +28,14 @@ public class ContainerTokenPBImpl extends ProtoBase<ContainerTokenProto> impleme
     viaProto = true;
   }
   
-  public ContainerTokenProto getProto() {
+  public synchronized ContainerTokenProto getProto() {
       mergeLocalToProto();
     proto = viaProto ? proto : builder.build();
     viaProto = true;
     return proto;
   }
 
-  private void mergeLocalToBuilder() {
+  private synchronized void mergeLocalToBuilder() {
     if (this.identifier != null) {
       builder.setIdentifier(convertToProtoFormat(this.identifier));
     }
@@ -44,7 +44,7 @@ public class ContainerTokenPBImpl extends ProtoBase<ContainerTokenProto> impleme
     }
   }
 
-  private void mergeLocalToProto() {
+  private synchronized void mergeLocalToProto() {
     if (viaProto) 
       maybeInitBuilder();
     mergeLocalToBuilder();
@@ -52,7 +52,7 @@ public class ContainerTokenPBImpl extends ProtoBase<ContainerTokenProto> impleme
     viaProto = true;
   }
 
-  private void maybeInitBuilder() {
+  private synchronized void maybeInitBuilder() {
     if (viaProto || builder == null) {
       builder = ContainerTokenProto.newBuilder(proto);
     }
@@ -61,7 +61,7 @@ public class ContainerTokenPBImpl extends ProtoBase<ContainerTokenProto> impleme
     
   
   @Override
-  public ByteBuffer getIdentifier() {
+  public synchronized ByteBuffer getIdentifier() {
     ContainerTokenProtoOrBuilder p = viaProto ? proto : builder;
     if (this.identifier != null) {
       return this.identifier;
@@ -74,14 +74,14 @@ public class ContainerTokenPBImpl extends ProtoBase<ContainerTokenProto> impleme
   }
 
   @Override
-  public void setIdentifier(ByteBuffer identifier) {
+  public synchronized void setIdentifier(ByteBuffer identifier) {
     maybeInitBuilder();
     if (identifier == null) 
       builder.clearIdentifier();
     this.identifier = identifier;
   }
   @Override
-  public ByteBuffer getPassword() {
+  public synchronized ByteBuffer getPassword() {
     ContainerTokenProtoOrBuilder p = viaProto ? proto : builder;
     if (this.password != null) {
       return this.password;
@@ -94,14 +94,14 @@ public class ContainerTokenPBImpl extends ProtoBase<ContainerTokenProto> impleme
   }
 
   @Override
-  public void setPassword(ByteBuffer password) {
+  public synchronized void setPassword(ByteBuffer password) {
     maybeInitBuilder();
     if (password == null) 
       builder.clearPassword();
     this.password = password;
   }
   @Override
-  public String getKind() {
+  public synchronized String getKind() {
     ContainerTokenProtoOrBuilder p = viaProto ? proto : builder;
     if (!p.hasKind()) {
       return null;
@@ -110,7 +110,7 @@ public class ContainerTokenPBImpl extends ProtoBase<ContainerTokenProto> impleme
   }
 
   @Override
-  public void setKind(String kind) {
+  public synchronized void setKind(String kind) {
     maybeInitBuilder();
     if (kind == null) {
       builder.clearKind();
@@ -119,7 +119,7 @@ public class ContainerTokenPBImpl extends ProtoBase<ContainerTokenProto> impleme
     builder.setKind((kind));
   }
   @Override
-  public String getService() {
+  public synchronized String getService() {
     ContainerTokenProtoOrBuilder p = viaProto ? proto : builder;
     if (!p.hasService()) {
       return null;
@@ -128,7 +128,7 @@ public class ContainerTokenPBImpl extends ProtoBase<ContainerTokenProto> impleme
   }
 
   @Override
-  public void setService(String service) {
+  public synchronized void setService(String service) {
     maybeInitBuilder();
     if (service == null) {
       builder.clearService();

mapreduce/yarn/yarn-api/src/main/proto/yarn_service_protos.proto

@@ -33,7 +33,7 @@ message FinishApplicationMasterResponseProto {
 message AllocateRequestProto {
   optional ApplicationAttemptIdProto application_attempt_id = 1;
   repeated ResourceRequestProto ask = 2;
-  repeated ContainerProto release = 3;
+  repeated ContainerIdProto release = 3;
   optional int32 response_id = 4;
   optional float progress = 5;
 }

mapreduce/yarn/yarn-server/yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/ApplicationMasterService.java

@@ -40,7 +40,7 @@ import org.apache.hadoop.yarn.api.protocolrecords.RegisterApplicationMasterReque
 import org.apache.hadoop.yarn.api.protocolrecords.RegisterApplicationMasterResponse;
 import org.apache.hadoop.yarn.api.records.AMResponse;
 import org.apache.hadoop.yarn.api.records.ApplicationAttemptId;
-import org.apache.hadoop.yarn.api.records.Container;
+import org.apache.hadoop.yarn.api.records.ContainerId;
 import org.apache.hadoop.yarn.api.records.ResourceRequest;
 import org.apache.hadoop.yarn.conf.YarnConfiguration;
 import org.apache.hadoop.yarn.exceptions.YarnRemoteException;
@@ -213,7 +213,7 @@ public class ApplicationMasterService extends AbstractService implements
               .getProgress()));
 
       List<ResourceRequest> ask = request.getAskList();
-      List<Container> release = request.getReleaseList();
+      List<ContainerId> release = request.getReleaseList();
 
       // Send new requests to appAttempt.
       Allocation allocation = 

mapreduce/yarn/yarn-server/yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/rmapp/attempt/RMAppAttemptImpl.java

@@ -15,6 +15,7 @@ import org.apache.commons.logging.LogFactory;
 import org.apache.hadoop.yarn.api.records.ApplicationAttemptId;
 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.NodeId;
 import org.apache.hadoop.yarn.api.records.Priority;
 import org.apache.hadoop.yarn.api.records.ResourceRequest;
@@ -421,8 +422,8 @@ public class RMAppAttemptImpl implements RMAppAttempt {
     }
   }
 
-  private static final List<Container> EMPTY_CONTAINER_RELEASE_LIST = 
-      new ArrayList<Container>();
+  private static final List<ContainerId> EMPTY_CONTAINER_RELEASE_LIST = 
+      new ArrayList<ContainerId>();
   private static final List<ResourceRequest> EMPTY_CONTAINER_REQUEST_LIST = 
     new ArrayList<ResourceRequest>();
 

mapreduce/yarn/yarn-server/yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/rmcontainer/ContainerAllocationExpirer.java

@@ -20,7 +20,6 @@ package org.apache.hadoop.yarn.server.resourcemanager.rmcontainer;
 
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.yarn.SystemClock;
-import org.apache.hadoop.yarn.api.records.Container;
 import org.apache.hadoop.yarn.api.records.ContainerId;
 import org.apache.hadoop.yarn.event.Dispatcher;
 import org.apache.hadoop.yarn.event.EventHandler;
@@ -29,7 +28,7 @@ import org.apache.hadoop.yarn.server.resourcemanager.scheduler.event.ContainerEx
 import org.apache.hadoop.yarn.util.AbstractLivelinessMonitor;
 
 public class ContainerAllocationExpirer extends
-    AbstractLivelinessMonitor<Container> {
+    AbstractLivelinessMonitor<ContainerId> {
 
   private EventHandler dispatcher;
 
@@ -48,7 +47,7 @@ public class ContainerAllocationExpirer extends
   }
 
   @Override
-  protected void expire(Container container) {
-    dispatcher.handle(new ContainerExpiredSchedulerEvent(container));
+  protected void expire(ContainerId containerId) {
+    dispatcher.handle(new ContainerExpiredSchedulerEvent(containerId));
   }
 }

mapreduce/yarn/yarn-server/yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/rmcontainer/RMContainerImpl.java

@@ -233,7 +233,7 @@ public class RMContainerImpl implements RMContainer {
     @Override
     public void transition(RMContainerImpl container, RMContainerEvent event) {
       // Register with containerAllocationExpirer.
-      container.containerAllocationExpirer.register(container.getContainer());
+      container.containerAllocationExpirer.register(container.getContainerId());
 
       // Tell the appAttempt
       container.eventHandler.handle(new RMAppAttemptContainerAcquiredEvent(
@@ -246,7 +246,8 @@ public class RMContainerImpl implements RMContainer {
     @Override
     public void transition(RMContainerImpl container, RMContainerEvent event) {
       // Unregister from containerAllocationExpirer.
-      container.containerAllocationExpirer.unregister(container.getContainer());
+      container.containerAllocationExpirer.unregister(container
+          .getContainerId());
     }
   }
 
@@ -267,7 +268,8 @@ public class RMContainerImpl implements RMContainer {
     public void transition(RMContainerImpl container, RMContainerEvent event) {
 
       // Unregister from containerAllocationExpirer.
-      container.containerAllocationExpirer.unregister(container.getContainer());
+      container.containerAllocationExpirer.unregister(container
+          .getContainerId());
 
       // Inform AppAttempt
       super.transition(container, event);
@@ -280,7 +282,8 @@ public class RMContainerImpl implements RMContainer {
     public void transition(RMContainerImpl container, RMContainerEvent event) {
 
       // Unregister from containerAllocationExpirer.
-      container.containerAllocationExpirer.unregister(container.getContainer());
+      container.containerAllocationExpirer.unregister(container
+          .getContainerId());
 
       // Inform node
       container.eventHandler.handle(new RMNodeCleanContainerEvent(

mapreduce/yarn/yarn-server/yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/YarnScheduler.java

@@ -25,7 +25,7 @@ import org.apache.hadoop.classification.InterfaceAudience.Private;
 import org.apache.hadoop.classification.InterfaceAudience.Public;
 import org.apache.hadoop.classification.InterfaceStability.Stable;
 import org.apache.hadoop.yarn.api.records.ApplicationAttemptId;
-import org.apache.hadoop.yarn.api.records.Container;
+import org.apache.hadoop.yarn.api.records.ContainerId;
 import org.apache.hadoop.yarn.api.records.NodeId;
 import org.apache.hadoop.yarn.api.records.QueueInfo;
 import org.apache.hadoop.yarn.api.records.QueueUserACLInfo;
@@ -94,7 +94,7 @@ public interface YarnScheduler extends EventHandler<SchedulerEvent> {
   Allocation 
   allocate(ApplicationAttemptId appAttemptId, 
       List<ResourceRequest> ask,
-      List<Container> release);
+      List<ContainerId> release);
 
   /**
    * Get node resource usage report.

mapreduce/yarn/yarn-server/yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/CapacityScheduler.java

@@ -389,7 +389,7 @@ implements ResourceScheduler, CapacitySchedulerContext {
   @Override
   @Lock(Lock.NoLock.class)
   public Allocation allocate(ApplicationAttemptId applicationAttemptId,
-      List<ResourceRequest> ask, List<Container> release) {
+      List<ResourceRequest> ask, List<ContainerId> release) {
 
     SchedulerApp application = getApplication(applicationAttemptId);
     if (application == null) {
@@ -402,8 +402,8 @@ implements ResourceScheduler, CapacitySchedulerContext {
     normalizeRequests(ask);
 
     // Release containers
-    for (Container releasedContainer : release) {
-      completedContainer(getRMContainer(releasedContainer), 
+    for (ContainerId releasedContainerId : release) {
+      completedContainer(getRMContainer(releasedContainerId), 
           RMContainerEventType.RELEASED);
     }
 
@@ -492,7 +492,7 @@ implements ResourceScheduler, CapacitySchedulerContext {
           containerLaunchedOnNode(container, node);
         } else { // has to be 'COMPLETE'
           LOG.info("DEBUG --- Container FINISHED: " + container.getId());
-          completedContainer(getRMContainer(container), 
+          completedContainer(getRMContainer(container.getId()), 
               RMContainerEventType.FINISHED);
         }
       }
@@ -588,7 +588,7 @@ implements ResourceScheduler, CapacitySchedulerContext {
     {
       ContainerExpiredSchedulerEvent containerExpiredEvent = 
           (ContainerExpiredSchedulerEvent) event;
-      completedContainer(getRMContainer(containerExpiredEvent.getContainer()), 
+      completedContainer(getRMContainer(containerExpiredEvent.getContainerId()), 
           RMContainerEventType.EXPIRE);
     }
     break;
@@ -671,10 +671,9 @@ implements ResourceScheduler, CapacitySchedulerContext {
     return nodes.get(nodeId);
   }
 
-  private RMContainer getRMContainer(Container container) {
-    ContainerId containerId = container.getId();
+  private RMContainer getRMContainer(ContainerId containerId) {
     SchedulerApp application = 
-        getApplication(container.getId().getAppAttemptId());
+        getApplication(containerId.getAppAttemptId());
     return (application == null) ? null : application.getRMContainer(containerId);
   }
 

mapreduce/yarn/yarn-server/yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/event/ContainerExpiredSchedulerEvent.java

@@ -1,24 +1,24 @@
 package org.apache.hadoop.yarn.server.resourcemanager.scheduler.event;
 
-import org.apache.hadoop.yarn.api.records.Container;
+import org.apache.hadoop.yarn.api.records.ContainerId;
 import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.ContainerAllocationExpirer;
 
 /**
- * The {@link SchedulerEvent} which notifies that a {@link Container}
+ * The {@link SchedulerEvent} which notifies that a {@link ContainerId}
  * has expired, sent by {@link ContainerAllocationExpirer} 
  *
  */
 public class ContainerExpiredSchedulerEvent extends SchedulerEvent {
 
-  private final Container container;
+  private final ContainerId containerId;
   
-  public ContainerExpiredSchedulerEvent(Container container) {
+  public ContainerExpiredSchedulerEvent(ContainerId containerId) {
     super(SchedulerEventType.CONTAINER_EXPIRED);
-    this.container = container;
+    this.containerId = containerId;
   }
 
-  public Container getContainer() {
-    return container;
+  public ContainerId getContainerId() {
+    return containerId;
   }
 
 }

mapreduce/yarn/yarn-server/yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fifo/FifoScheduler.java

@@ -212,7 +212,7 @@ public class FifoScheduler implements ResourceScheduler {
   @Override
   public Allocation allocate(
       ApplicationAttemptId applicationAttemptId, List<ResourceRequest> ask,
-      List<Container> release) {
+      List<ContainerId> release) {
     SchedulerApp application = getApplication(applicationAttemptId);
     if (application == null) {
       LOG.error("Calling allocate on removed " +
@@ -224,7 +224,7 @@ public class FifoScheduler implements ResourceScheduler {
     normalizeRequests(ask);
 
     // Release containers
-    for (Container releasedContainer : release) {
+    for (ContainerId releasedContainer : release) {
       containerCompleted(getRMContainer(releasedContainer), 
           RMContainerEventType.RELEASED);
     }
@@ -543,7 +543,7 @@ public class FifoScheduler implements ResourceScheduler {
         if (container.getState() == ContainerState.RUNNING) {
           containerLaunchedOnNode(container, node);
         } else { // has to COMPLETE
-          containerCompleted(getRMContainer(container), 
+          containerCompleted(getRMContainer(container.getId()), 
               RMContainerEventType.FINISHED);
         }
       }
@@ -610,7 +610,7 @@ public class FifoScheduler implements ResourceScheduler {
     {
       ContainerExpiredSchedulerEvent containerExpiredEvent = 
           (ContainerExpiredSchedulerEvent) event;
-      containerCompleted(getRMContainer(containerExpiredEvent.getContainer()), 
+      containerCompleted(getRMContainer(containerExpiredEvent.getContainerId()), 
           RMContainerEventType.EXPIRE);
     }
     break;
@@ -716,10 +716,9 @@ public class FifoScheduler implements ResourceScheduler {
         node.getUsedResource(), node.getNumContainers());
   }
   
-  private RMContainer getRMContainer(Container container) {
-    ContainerId containerId = container.getId();
+  private RMContainer getRMContainer(ContainerId containerId) {
     SchedulerApp application = 
-        getApplication(container.getId().getAppAttemptId());
+        getApplication(containerId.getAppAttemptId());
     return application.getRMContainer(containerId);
   }
 

mapreduce/yarn/yarn-server/yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/Application.java

@@ -266,7 +266,7 @@ public class Application {
     
     // Get resources from the ResourceManager
     resourceManager.getResourceScheduler().allocate(applicationAttemptId,
-        new ArrayList<ResourceRequest>(ask), new ArrayList<Container>());
+        new ArrayList<ResourceRequest>(ask), new ArrayList<ContainerId>());
     System.out.println("-=======" + applicationAttemptId);
     System.out.println("----------" + resourceManager.getRMContext().getRMApps()
         .get(applicationId).getRMAppAttempt(applicationAttemptId));

mapreduce/yarn/yarn-server/yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/MockAM.java

@@ -30,7 +30,7 @@ public class MockAM {
   private final AMRMProtocol amRMProtocol;
   
   private final List<ResourceRequest> requests = new ArrayList<ResourceRequest>();
-  private final List<Container> releases = new ArrayList<Container>();
+  private final List<ContainerId> releases = new ArrayList<ContainerId>();
 
   MockAM(RMContext context, AMRMProtocol amRMProtocol, 
       ApplicationAttemptId attemptId) {
@@ -82,13 +82,7 @@ public class MockAM {
       String host, int memory, int numContainers, 
       List<ContainerId> releases) throws Exception {
     List reqs = createReq(new String[]{host}, memory, 1, numContainers);
-    List<Container> toRelease = new ArrayList<Container>();
-    for (ContainerId id : releases) {
-      Container cont = Records.newRecord(Container.class);
-      cont.setId(id);
-      //TOOD: set all fields
-    }
-    return allocate(reqs, toRelease);
+    return allocate(reqs, releases);
   }
 
   public List<ResourceRequest> createReq(String[] hosts, int memory, int priority, 
@@ -125,7 +119,7 @@ public class MockAM {
   }
 
   public AMResponse allocate(
-      List<ResourceRequest> resourceRequest, List<Container> releases) 
+      List<ResourceRequest> resourceRequest, List<ContainerId> releases) 
       throws Exception {
     AllocateRequest req = Records.newRecord(AllocateRequest.class);
     req.setResponseId(++responseId);

mapreduce/yarn/yarn-server/yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/TestApplicationCleanup.java

@@ -48,11 +48,11 @@ public class TestApplicationCleanup {
     //kick the scheduler
     nm1.nodeHeartbeat(true);
     List<Container> conts = am.allocate(new ArrayList<ResourceRequest>(),
-        new ArrayList<Container>()).getNewContainerList();
+        new ArrayList<ContainerId>()).getNewContainerList();
     int contReceived = conts.size();
     while (contReceived < request) {
       conts = am.allocate(new ArrayList<ResourceRequest>(),
-          new ArrayList<Container>()).getNewContainerList();
+          new ArrayList<ContainerId>()).getNewContainerList();
       contReceived += conts.size();
       Log.info("Got " + contReceived + " containers. Waiting to get " + request);
       Thread.sleep(2000);

mapreduce/yarn/yarn-server/yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/TestRM.java

@@ -68,11 +68,11 @@ public class TestRM {
     //kick the scheduler
     nm1.nodeHeartbeat(true);
     List<Container> conts = am.allocate(new ArrayList<ResourceRequest>(),
-        new ArrayList<Container>()).getNewContainerList();
+        new ArrayList<ContainerId>()).getNewContainerList();
     int contReceived = conts.size();
     while (contReceived < 3) {//only 3 containers are available on node1
       conts.addAll(am.allocate(new ArrayList<ResourceRequest>(),
-          new ArrayList<Container>()).getNewContainerList());
+          new ArrayList<ContainerId>()).getNewContainerList());
       contReceived = conts.size();
       LOG.info("Got " + contReceived + " containers. Waiting to get " + 3);
       Thread.sleep(2000);
@@ -82,11 +82,11 @@ public class TestRM {
     //send node2 heartbeat
     nm2.nodeHeartbeat(true);
     conts = am.allocate(new ArrayList<ResourceRequest>(),
-        new ArrayList<Container>()).getNewContainerList();
+        new ArrayList<ContainerId>()).getNewContainerList();
     contReceived = conts.size();
     while (contReceived < 10) {
       conts.addAll(am.allocate(new ArrayList<ResourceRequest>(),
-          new ArrayList<Container>()).getNewContainerList());
+          new ArrayList<ContainerId>()).getNewContainerList());
       contReceived = conts.size();
       LOG.info("Got " + contReceived + " containers. Waiting to get " + 10);
       Thread.sleep(2000);

mapreduce/yarn/yarn-server/yarn-server-tests/src/test/java/org/apache/hadoop/yarn/server/TestContainerTokenSecretManager.java

@@ -231,7 +231,7 @@ public class TestContainerTokenSecretManager {
     rr.setPriority(recordFactory.newRecordInstance(Priority.class));
     rr.getPriority().setPriority(0);
     ask.add(rr);
-    ArrayList<Container> release = new ArrayList<Container>();
+    ArrayList<ContainerId> release = new ArrayList<ContainerId>();
     
     AllocateRequest allocateRequest =
         recordFactory.newRecordInstance(AllocateRequest.class);