package com.hubspot.singularity.scheduler;

import java.util.Iterator;
import java.util.List;
import java.util.Set;

import org.apache.mesos.Protos.TaskStatus;
import org.apache.mesos.SchedulerDriver;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.google.common.collect.Sets;
import com.google.inject.Inject;
import com.hubspot.singularity.SingularityTaskId;
import com.hubspot.singularity.data.TaskManager;

public class SingularityTaskReconciliation {

  private static final Logger LOG = LoggerFactory.getLogger(SingularityTaskReconciliation.class);

  private final TaskManager taskManager;

  @Inject
  public SingularityTaskReconciliation(TaskManager taskManager) {
    this.taskManager = taskManager;
  }

  public void reconcileTasks(SchedulerDriver driver) {
    Set<SingularityTaskId> taskIds = Sets.newHashSet(taskManager.getActiveTaskIds());
    List<TaskStatus> taskStatuses = taskManager.getLastActiveTaskStatuses();

    for (Iterator<TaskStatus> taskStatusItr = taskStatuses.iterator(); taskStatusItr.hasNext();) {
      TaskStatus taskStatus = taskStatusItr.next();
      SingularityTaskId taskId = SingularityTaskId.fromString(taskStatus.getTaskId().getValue());

      if (!taskIds.contains(taskId)) {
        LOG.info("Task {} not found, deleting taskStatus", taskId);
        taskManager.deleteLastActiveTaskStatus(taskId);
        taskStatusItr.remove();
      }
    }

    driver.reconcileTasks(taskStatuses);
  }

}

@@ -110,16 +110,55 @@
 l.Set;%0A%0A
+import org.apache.mesos.Protos.Status;%0A
 import o
@@ -184,32 +184,32 @@
 tos.TaskStatus;%0A
-
 import org.apach
@@ -363,16 +363,52 @@
 Inject;%0A
+import com.hubspot.mesos.JavaUtils;%0A
 import c
@@ -854,24 +854,76 @@
 r driver) %7B%0A
+    final long start = System.currentTimeMillis();%0A%0A
     Set%3CSing
@@ -1542,42 +1542,289 @@
     
-driver.reconcileTasks(taskStatuses
+LOG.info(%22Requesting reconciliation for %7B%7D taskStatuses%22, taskStatuses.size());%0A%0A    Status status = driver.reconcileTasks(taskStatuses);%0A%0A    LOG.info(%22Requested reconciliation of %7B%7D taskStatuses with driver status %7B%7D in %7B%7D%22, taskStatuses.size(), status, JavaUtils.duration(start)
 );%0A 

add logging to reconcile