Addressed review comments

master/src/worker_management/state.rs

@@ -55,11 +55,9 @@ impl State {
     fn get_in_progress_tasks_for_job(&self, job_id: &str) -> Result<Vec<String>> {
         let mut tasks: Vec<String> = Vec::new();
 
-        for task_id in self.tasks.keys() {
-            if let Some(task) = self.tasks.get(&task_id.clone()) {
-                if task.job_id == job_id {
-                    tasks.push(task_id.clone());
-                }
+        for task in self.tasks.values() {
+            if task.job_id == job_id {
+                tasks.push(task.id.clone());
             }
         }
 
@@ -515,23 +513,17 @@ impl State {
             new_map_task.job_priority * FAILED_TASK_PRIORITY,
         ));
 
-        println!("Rescheduled map task with ID {}", new_map_task.id);
+        info!("Rescheduled map task with ID {}", new_map_task.id);
         Ok(())
     }
 
     pub fn handle_worker_report(&mut self, request: &pb::ReportWorkerRequest) -> Result<()> {
         let mut reschedule_task = String::new();
 
         if self.tasks.contains_key(&request.task_id) {
-            let task_ids = self.completed_tasks.keys().clone();
-
-            for task_id in task_ids {
-                let map_task = self.completed_tasks.get(task_id).chain_err(|| {
-                    format!("Unable to get map task with ID {}", task_id)
-                })?;
-
-                if self.path_from_map_task(map_task, &request.path) {
-                    reschedule_task = map_task.id.clone();
+            for task in self.completed_tasks.values() {
+                if self.path_from_map_task(task, &request.path) {
+                    reschedule_task = task.id.clone();
                     break;
                 }