 +        // 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<TaskDescription>> replies =
-+            new ArrayList<List<TaskDescription>>(numOffers);
++        List<List<TaskInfo>> replies =
++            new ArrayList<List<TaskInfo>>(numOffers);
 +        for (int i = 0; i < numOffers; i++) {
 +          indices.add(i);
-+          replies.add(new ArrayList<TaskDescription>());
++          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);
-+            TaskDescription task = findTask(
++            TaskInfo task = findTask(
 +                offer.getSlaveId(), offer.getHostname(), cpus[i], mem[i]);
 +            if (task != null) {
 +              cpus[i] -= getResource(task, "cpus");

+             " as " + taskType + " on " + host);
+    
+    // Create a task description to pass back to Mesos.
+    return TaskDescription.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.TaskInfo;
+import org.apache.mesos.Protos.TaskState;
+import org.apache.mesos.Protos.Status;
+import org.apache.mesos.Scheduler;
