 +             " as " + taskType + " on " + host);
 +    
 +    // Create a task description to pass back to Mesos.
-+    return TaskDescription.newBuilder()
++    return TaskInfo.newBuilder()
 +        .setTaskId(mesosId)
 +        .setSlaveId(slaveId)
 +        .setName("task " + mesosId.getValue() + " (" + taskType + ")")

+import org.apache.mesos.Protos.SlaveID;
+import org.apache.mesos.Protos.Offer;
+import org.apache.mesos.Protos.TaskID;
+import org.apache.mesos.Protos.TaskDescription;
+import org.apache.mesos.Protos.TaskState;
+import org.apache.mesos.Protos.Status;
+import org.apache.mesos.Scheduler;

+        // 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");
