package com.twitter.mesos.scheduler;

import java.util.logging.Level;
import java.util.logging.Logger;

import com.google.common.base.Optional;
import com.google.common.collect.ImmutableList;
import com.google.inject.Inject;

import org.apache.mesos.Protos.Offer;
import org.apache.mesos.Protos.OfferID;
import org.apache.mesos.Protos.TaskInfo;
import org.apache.mesos.Protos.TaskStatus;

import com.twitter.mesos.StateTranslator;
import com.twitter.mesos.gen.ScheduleStatus;
import com.twitter.mesos.gen.TaskQuery;
import com.twitter.mesos.scheduler.async.TaskScheduler;

import static com.google.common.base.Preconditions.checkNotNull;

/**
 * A task launcher that matches resource offers against user tasks.
 */
class UserTaskLauncher implements TaskLauncher {

  private static final Logger LOG = Logger.getLogger(UserTaskLauncher.class.getName());

  private final TaskScheduler taskScheduler;
  private final StateManager stateManager;

  @Inject
  UserTaskLauncher(TaskScheduler taskScheduler, StateManager stateManager) {
    this.taskScheduler = checkNotNull(taskScheduler);
    this.stateManager = checkNotNull(stateManager);
  }

  @Override
  public Optional<TaskInfo> createTask(Offer offer) {
    checkNotNull(offer);

    stateManager.saveAttributesFromOffer(offer.getHostname(), offer.getAttributesList());
    taskScheduler.offer(ImmutableList.of(offer));

    return Optional.absent();
  }

  @Override
  public synchronized boolean statusUpdate(TaskStatus status) {
    String info = status.hasData() ? status.getData().toStringUtf8() : null;
    TaskQuery query = Query.byId(status.getTaskId().getValue());

    try {
      ScheduleStatus translatedState = StateTranslator.get(status.getState());
      if (translatedState == null) {
        LOG.severe("Failed to look up task state translation for: " + status.getState());
      } else {
        stateManager.changeState(query, translatedState, Optional.fromNullable(info));
        return true;
      }
    } catch (SchedulerException e) {
      LOG.log(Level.WARNING, "Failed to update status for: " + status, e);
      throw e;
    }
    return false;
  }

  @Override
  public void cancelOffer(OfferID offer) {
    taskScheduler.cancelOffer(offer);
  }
}

@@ -93,24 +93,59 @@
 ng.Logger;%0A%0A
+import javax.annotation.Nullable;%0A%0A
 import com.g
@@ -1521,40 +1521,254 @@
     
-String info = status.hasData() ?
+@Nullable String message = null;%0A    if (status.hasMessage()) %7B%0A      message = status.getMessage();%0A    %7D else if (status.hasData()) %7B%0A      // TODO(William Farner): Once thermos executor has full deployment, this may be removed.%0A      message =
  sta
@@ -1799,16 +1799,16 @@
 f8()
- : null;
+;%0A    %7D%0A
 %0A   
@@ -2184,12 +2184,15 @@
 ble(
-info
+message
 ));%0A

Use status update message field by default.