package com.twitter.mesos.scheduler;

import java.util.logging.Logger;

import javax.annotation.Nullable;
import javax.inject.Provider;

import com.google.common.base.Function;
import com.google.common.base.Preconditions;
import com.google.inject.Inject;

import org.apache.mesos.MesosSchedulerDriver;
import org.apache.mesos.Protos.FrameworkID;
import org.apache.mesos.Protos.FrameworkInfo;
import org.apache.mesos.Scheduler;
import org.apache.mesos.SchedulerDriver;

import com.twitter.common.args.Arg;
import com.twitter.common.args.CmdLine;
import com.twitter.common.args.constraints.NotNull;
import com.twitter.common.quantity.Amount;
import com.twitter.common.quantity.Time;

/**
 * Factory to create scheduler driver instances.
 */
public interface DriverFactory extends Function<String, SchedulerDriver> {

  static class DriverFactoryImpl implements DriverFactory {
    private static final Logger LOG = Logger.getLogger(DriverFactory.class.getName());

    @NotNull
    @CmdLine(name = "mesos_master_address",
            help = "Mesos address for the master, can be a mesos address or zookeeper path.")
    private static final Arg<String> MESOS_MASTER_ADDRESS = Arg.create();

    @CmdLine(name = "framework_failover_timeout",
        help = "Time after which a framework is considered deleted.  SHOULD BE VERY HIGH.")
    private static final Arg<Amount<Long, Time>> FRAMEWORK_FAILOVER_TIMEOUT =
        Arg.create(Amount.of(21L, Time.DAYS));

    private static final String EXECUTOR_USER = "root";

    private static final String TWITTER_FRAMEWORK_NAME = "TwitterScheduler";

    private final Provider<Scheduler> scheduler;

    @Inject
    DriverFactoryImpl(Provider<Scheduler> scheduler) {
      this.scheduler = Preconditions.checkNotNull(scheduler);
    }

    @Override
    public SchedulerDriver apply(@Nullable String frameworkId) {
      LOG.info("Connecting to mesos master: " + MESOS_MASTER_ADDRESS.get());

      FrameworkInfo.Builder frameworkInfo = FrameworkInfo.newBuilder()
          .setUser(EXECUTOR_USER)
          .setName(TWITTER_FRAMEWORK_NAME)
          .setFailoverTimeout(FRAMEWORK_FAILOVER_TIMEOUT.get().as(Time.SECONDS));

      if (frameworkId != null) {
        LOG.info("Found persisted framework ID: " + frameworkId);
        frameworkInfo.setId(FrameworkID.newBuilder().setValue(frameworkId));
      } else {
        LOG.warning("Did not find a persisted framework ID, connecting as a new framework.");
      }

      return new MesosSchedulerDriver(scheduler.get(), frameworkInfo.build(),
          MESOS_MASTER_ADDRESS.get());
    }
  }
}

@@ -1446,24 +1446,1621 @@
 me.DAYS));%0A%0A
+    /**%0A     * Require Mesos slaves to have checkpointing enabled. Slaves with checkpointing enabled will%0A     * attempt to write checkpoints when required by a task's framework. These checkpoints allow%0A     * executors to be reattached rather than killed when a slave is restarted.%0A     *%0A     * This flag is dangerous! When enabled tasks will not launch on slaves without checkpointing%0A     * enabled.%0A     *%0A     * Behavior is as follows:%0A     * (Scheduler -require_slave_checkpoint=true,  Slave --checkpoint=true):%0A     *   Tasks will launch.        Checkpoints will be written.%0A     * (Scheduler -require_slave_checkpoint=true,   Slave --checkpoint=false):%0A     *   Tasks WILL NOT launch.%0A     * (Scheduler -require_slave_checkpoint=false,  Slave --checkpoint=true):%0A     *   Tasks will launch.        Checkpoints will not be written.%0A     * (Scheduler -require_slave_checkpoint=false,  Slave --checkpoint=false):%0A     *   Tasks will launch.        Checkpoints will not be written.%0A     *%0A     * TODO(ksweeney): Remove warning table after https://issues.apache.org/jira/browse/MESOS-444%0A     * is resolved.%0A     */%0A    @CmdLine(name = %22require_slave_checkpoint%22,%0A        help = %22DANGEROUS! Require Mesos slaves to have checkpointing enabled. When enabled a %22%0A            + %22slave restart should not kill executors, but the scheduler will not be able to %22%0A            + %22launch tasks on slaves without --checkpoint=true in their command lines. See %22%0A            + %22DriverFactory.java for more information.%22)%0A    private static final Arg%3CBoolean%3E REQUIRE_SLAVE_CHECKPOINT = Arg.create(false);%0A%0A
     private 
@@ -3673,16 +3673,73 @@
 K_NAME)%0A
+          .setCheckpoint(REQUIRE_SLAVE_CHECKPOINT.get())%0A
         

Bump Aurora dependency to mesos-0.13.0. Add -require_slave_checkpoint.