 +import org.apache.mesos.Protos.SlaveInfo;
 +import org.apache.mesos.Protos.Status;
 +import org.apache.mesos.Protos.TaskID;
-+import org.apache.mesos.Protos.TaskDescription;
++import org.apache.mesos.Protos.TaskInfo;
 +import org.apache.mesos.Protos.TaskState;
 +
 +public class FrameworkExecutor implements Executor {

+    return getResource(offer.getResourcesList(), name);
+  }
+
+  private static double getResource(TaskDescription task, String name) {
+    return getResource(task.getResourcesList(), name);
+  }
+  

+        // because it minimizing the amount of scanning we need to do if we
+        // get a large set of offered nodes.
+        List<Integer> indices = new LinkedList<Integer>();
+        List<List<TaskInfo>> replies =
+            new ArrayList<List<TaskInfo>>(numOffers);
+        for (int i = 0; i < numOffers; i++) {
+          indices.add(i);
+          replies.add(new ArrayList<TaskInfo>());
+        }
+        while (indices.size() > 0) {
+          for (Iterator<Integer> it = indices.iterator(); it.hasNext();) {
+            int i = it.next();
+            Offer offer = offers.get(i);
+            TaskInfo task = findTask(
+                offer.getSlaveId(), offer.getHostname(), cpus[i], mem[i]);
+            if (task != null) {
+              cpus[i] -= getResource(task, "cpus");
