/**
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.apache.aurora.scheduler.updater;

import java.util.Collection;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
import java.util.concurrent.ScheduledExecutorService;
import java.util.stream.Collectors;

import com.google.common.base.Function;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.ImmutableMultimap;
import com.google.common.collect.ImmutableSet;
import com.google.common.collect.Iterables;
import com.google.common.collect.Maps;
import com.google.common.collect.Multimap;
import com.google.common.collect.Multimaps;
import com.google.common.collect.Ordering;
import com.google.common.eventbus.EventBus;
import com.google.common.primitives.Ints;
import com.google.inject.AbstractModule;
import com.google.inject.Guice;
import com.google.inject.Injector;

import org.apache.aurora.common.application.Lifecycle;
import org.apache.aurora.common.base.Command;
import org.apache.aurora.common.quantity.Amount;
import org.apache.aurora.common.quantity.Time;
import org.apache.aurora.common.stats.Stats;
import org.apache.aurora.common.stats.StatsProvider;
import org.apache.aurora.common.testing.easymock.EasyMockTest;
import org.apache.aurora.common.util.Clock;
import org.apache.aurora.common.util.TruncatedBinaryBackoff;
import org.apache.aurora.gen.BatchJobUpdateStrategy;
import org.apache.aurora.gen.CountSlaPolicy;
import org.apache.aurora.gen.InstanceTaskConfig;
import org.apache.aurora.gen.JobUpdate;
import org.apache.aurora.gen.JobUpdateAction;
import org.apache.aurora.gen.JobUpdateEvent;
import org.apache.aurora.gen.JobUpdateInstructions;
import org.apache.aurora.gen.JobUpdateKey;
import org.apache.aurora.gen.JobUpdatePulseStatus;
import org.apache.aurora.gen.JobUpdateSettings;
import org.apache.aurora.gen.JobUpdateState;
import org.apache.aurora.gen.JobUpdateStatus;
import org.apache.aurora.gen.JobUpdateStrategy;
import org.apache.aurora.gen.JobUpdateSummary;
import org.apache.aurora.gen.Metadata;
import org.apache.aurora.gen.QueueJobUpdateStrategy;
import org.apache.aurora.gen.Range;
import org.apache.aurora.gen.ScheduleStatus;
import org.apache.aurora.gen.ScheduledTask;
import org.apache.aurora.gen.ServerInfo;
import org.apache.aurora.gen.SlaPolicy;
import org.apache.aurora.gen.TaskConfig;
import org.apache.aurora.gen.VariableBatchJobUpdateStrategy;
import org.apache.aurora.scheduler.SchedulerModule.TaskEventBatchWorker;
import org.apache.aurora.scheduler.TaskIdGenerator;
import org.apache.aurora.scheduler.TaskIdGenerator.TaskIdGeneratorImpl;
import org.apache.aurora.scheduler.TierModule;
import org.apache.aurora.scheduler.base.JobKeys;
import org.apache.aurora.scheduler.base.Query;
import org.apache.aurora.scheduler.base.TaskTestUtil;
import org.apache.aurora.scheduler.base.Tasks;
import org.apache.aurora.scheduler.config.types.TimeAmount;
import org.apache.aurora.scheduler.events.EventSink;
import org.apache.aurora.scheduler.events.PubsubEvent;
import org.apache.aurora.scheduler.mesos.Driver;
import org.apache.aurora.scheduler.scheduling.RescheduleCalculator;
import org.apache.aurora.scheduler.scheduling.RescheduleCalculator.RescheduleCalculatorImpl;
import org.apache.aurora.scheduler.sla.SlaModule;
import org.apache.aurora.scheduler.state.StateChangeResult;
import org.apache.aurora.scheduler.state.StateManager;
import org.apache.aurora.scheduler.state.StateManagerImpl;
import org.apache.aurora.scheduler.state.UUIDGenerator;
import org.apache.aurora.scheduler.state.UUIDGenerator.UUIDGeneratorImpl;
import org.apache.aurora.scheduler.storage.JobUpdateStore;
import org.apache.aurora.scheduler.storage.Storage;
import org.apache.aurora.scheduler.storage.entities.IInstanceTaskConfig;
import org.apache.aurora.scheduler.storage.entities.IJobInstanceUpdateEvent;
import org.apache.aurora.scheduler.storage.entities.IJobKey;
import org.apache.aurora.scheduler.storage.entities.IJobUpdate;
import org.apache.aurora.scheduler.storage.entities.IJobUpdateDetails;
import org.apache.aurora.scheduler.storage.entities.IJobUpdateEvent;
import org.apache.aurora.scheduler.storage.entities.IJobUpdateKey;
import org.apache.aurora.scheduler.storage.entities.IJobUpdateSummary;
import org.apache.aurora.scheduler.storage.entities.IScheduledTask;
import org.apache.aurora.scheduler.storage.entities.IServerInfo;
import org.apache.aurora.scheduler.storage.entities.ITaskConfig;
import org.apache.aurora.scheduler.storage.mem.MemStorageModule;
import org.apache.aurora.scheduler.testing.FakeScheduledExecutor;
import org.apache.aurora.scheduler.testing.FakeStatsProvider;
import org.apache.aurora.scheduler.updater.JobUpdateController.AuditData;
import org.apache.aurora.scheduler.updater.StateEvaluator.Failure;
import org.apache.aurora.scheduler.updater.UpdaterModule.UpdateActionBatchWorker;
import org.easymock.EasyMock;
import org.easymock.IExpectationSetters;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import static org.apache.aurora.gen.JobUpdateAction.INSTANCE_ROLLBACK_FAILED;
import static org.apache.aurora.gen.JobUpdateAction.INSTANCE_ROLLED_BACK;
import static org.apache.aurora.gen.JobUpdateAction.INSTANCE_ROLLING_BACK;
import static org.apache.aurora.gen.JobUpdateAction.INSTANCE_UPDATED;
import static org.apache.aurora.gen.JobUpdateAction.INSTANCE_UPDATE_FAILED;
import static org.apache.aurora.gen.JobUpdateAction.INSTANCE_UPDATING;
import static org.apache.aurora.gen.JobUpdateStatus.ABORTED;
import static org.apache.aurora.gen.JobUpdateStatus.ERROR;
import static org.apache.aurora.gen.JobUpdateStatus.ROLLED_BACK;
import static org.apache.aurora.gen.JobUpdateStatus.ROLLED_FORWARD;
import static org.apache.aurora.gen.JobUpdateStatus.ROLLING_BACK;
import static org.apache.aurora.gen.JobUpdateStatus.ROLLING_FORWARD;
import static org.apache.aurora.gen.JobUpdateStatus.ROLL_BACK_PAUSED;
import static org.apache.aurora.gen.JobUpdateStatus.ROLL_FORWARD_AWAITING_PULSE;
import static org.apache.aurora.gen.JobUpdateStatus.ROLL_FORWARD_PAUSED;
import static org.apache.aurora.gen.ScheduleStatus.ASSIGNED;
import static org.apache.aurora.gen.ScheduleStatus.FAILED;
import static org.apache.aurora.gen.ScheduleStatus.FINISHED;
import static org.apache.aurora.gen.ScheduleStatus.KILLED;
import static org.apache.aurora.gen.ScheduleStatus.KILLING;
import static org.apache.aurora.gen.ScheduleStatus.RUNNING;
import static org.apache.aurora.gen.ScheduleStatus.STARTING;
import static org.apache.aurora.scheduler.storage.Storage.MutateWork.NoResult;
import static org.apache.aurora.scheduler.testing.BatchWorkerUtil.expectBatchExecute;
import static org.apache.aurora.scheduler.updater.UpdateFactory.UpdateFactoryImpl.expandInstanceIds;
import static org.easymock.EasyMock.expectLastCall;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;

public class JobUpdaterIT extends EasyMockTest {

  private static final String USER = "user";
  private static final AuditData AUDIT = new AuditData(USER, Optional.of("message"));
  private static final IJobKey JOB = JobKeys.from("role", "env", "job1");
  private static final IJobUpdateKey UPDATE_ID =
      IJobUpdateKey.build(new JobUpdateKey(JOB.newBuilder(), "update_id"));
  private static final Amount<Long, Time> WATCH_TIMEOUT = Amount.of(2000L, Time.MILLISECONDS);
  private static final TimeAmount FLAPPING_THRESHOLD = new TimeAmount(1L, Time.MILLISECONDS);
  private static final Amount<Long, Time> ONE_DAY = Amount.of(1L, Time.DAYS);
  private static final Amount<Long, Time> ONE_HOUR = Amount.of(1L, Time.HOURS);
  private static final Amount<Long, Time> ONE_MINUTE = Amount.of(1L, Time.MINUTES);
  private static final ITaskConfig OLD_CONFIG =
      setExecutorData(TaskTestUtil.makeConfig(JOB), "olddata");
  private static final ITaskConfig NEW_CONFIG = setExecutorData(OLD_CONFIG, "newdata");
  private static final ITaskConfig SLA_AWARE_CONFIG =
      setCountSlaPolicy(setExecutorData(OLD_CONFIG, "sladata"), 2, 0);
  private static final long PULSE_TIMEOUT_MS = 10000;
  private static final ImmutableSet<Metadata> METADATA = ImmutableSet.of(
      new Metadata("k1", "v1"), new Metadata("k2", "v2"));

  private FakeScheduledExecutor clock;
  private JobUpdateController updater;
  private Driver driver;
  private EventBus eventBus;
  private Storage storage;
  private StateManager stateManager;
  private JobUpdateEventSubscriber subscriber;
  private Command shutdownCommand;

  private static ITaskConfig setExecutorData(ITaskConfig task, String executorData) {
    TaskConfig builder = task.newBuilder();
    builder.getExecutorConfig().setData(executorData);
    return ITaskConfig.build(builder);
  }

  private static ITaskConfig setCountSlaPolicy(ITaskConfig task, int count, int durationMs) {
    TaskConfig builder = task.newBuilder();
    SlaPolicy policy = SlaPolicy.countSlaPolicy(new CountSlaPolicy(count, durationMs));
    builder.setSlaPolicy(policy);
    return ITaskConfig.build(builder);
  }

  @Before
  public void setUp() throws Exception {
    // Avoid console spam due to stats registered multiple times.
    Stats.flush();
    ScheduledExecutorService executor = createMock(ScheduledExecutorService.class);
    clock = FakeScheduledExecutor.scheduleExecutor(executor);
    driver = createMock(Driver.class);
    shutdownCommand = createMock(Command.class);
    eventBus = new EventBus();
    TaskEventBatchWorker taskEventBatchWorker = createMock(TaskEventBatchWorker.class);
    UpdateActionBatchWorker updateActionBatchWorker = createMock(UpdateActionBatchWorker.class);

    UpdaterModule.Options updaterOptions = new UpdaterModule.Options();
    updaterOptions.enableAffinity = true;
    updaterOptions.slaAwareKillRetryMinDelay = new TimeAmount(
        WATCH_TIMEOUT.getValue(),
        WATCH_TIMEOUT.getUnit());
    updaterOptions.slaAwareKillRetryMaxDelay = new TimeAmount(
        WATCH_TIMEOUT.getValue(),
        WATCH_TIMEOUT.getUnit());
    SlaModule.Options slaOptions = new SlaModule.Options();
    slaOptions.minRequiredInstances = 3;

    Injector injector = Guice.createInjector(
        new UpdaterModule(executor, Optional.of(updateActionBatchWorker), updaterOptions),
        new SlaModule(slaOptions),
        new TierModule(TaskTestUtil.TIER_CONFIG),
        new MemStorageModule(),
        new AbstractModule() {
          @Override
          protected void configure() {
            bind(StatsProvider.class).toInstance(new FakeStatsProvider());
            bind(Clock.class).toInstance(clock);
            bind(StateManager.class).to(StateManagerImpl.class);
            bind(Driver.class).toInstance(driver);
            bind(TaskIdGenerator.class).to(TaskIdGeneratorImpl.class);
            bind(RescheduleCalculator.class).to(RescheduleCalculatorImpl.class);
            bind(RescheduleCalculatorImpl.RescheduleCalculatorSettings.class)
                .toInstance(new RescheduleCalculatorImpl.RescheduleCalculatorSettings(
                    new TruncatedBinaryBackoff(
                        Amount.of(1L, Time.SECONDS), Amount.of(1L, Time.MINUTES)),
                    FLAPPING_THRESHOLD,
                    new TimeAmount(1, Time.MINUTES)));
            bind(EventSink.class).toInstance(eventBus::post);
            bind(UUIDGenerator.class).to(UUIDGeneratorImpl.class);
            bind(Lifecycle.class).toInstance(new Lifecycle(shutdownCommand));
            bind(TaskEventBatchWorker.class).toInstance(taskEventBatchWorker);
            bind(UpdateActionBatchWorker.class).toInstance(updateActionBatchWorker);
            bind(IServerInfo.class).toInstance(
                IServerInfo.build(
                    new ServerInfo()
                        .setClusterName("JobUpdaterITCluster")
                        .setStatsUrlPrefix("test_stats_prefix")));
          }
        });
    updater = injector.getInstance(JobUpdateController.class);
    storage = injector.getInstance(Storage.class);
    storage.prepare();
    stateManager = injector.getInstance(StateManager.class);
    eventBus.register(injector.getInstance(JobUpdateEventSubscriber.class));
    subscriber = injector.getInstance(JobUpdateEventSubscriber.class);
    expectBatchExecute(taskEventBatchWorker, storage, control).anyTimes();
    expectBatchExecute(updateActionBatchWorker, storage, control).anyTimes();
  }

  @After
  public void validateExitState() {
    clock.assertEmpty();
  }

  private String getTaskId(IJobKey job, int instanceId) {
    return Tasks.id(Iterables.getOnlyElement(
        Storage.Util.fetchTasks(
            storage,
            Query.instanceScoped(job, instanceId).active())));
  }

  private void changeState(
      IJobKey job,
      int instanceId,
      ScheduleStatus status,
      ScheduleStatus... statuses) {

    for (ScheduleStatus s
        : ImmutableList.<ScheduleStatus>builder().add(status).add(statuses).build()) {

      storage.write((NoResult.Quiet) storeProvider ->
          assertEquals(
              StateChangeResult.SUCCESS,
              stateManager.changeState(
                  storeProvider,
                  getTaskId(job, instanceId),
                  Optional.empty(),
                  s,
                  Optional.empty())));
    }
  }

  private static final Ordering<IJobInstanceUpdateEvent> EVENT_ORDER = Ordering.natural()
      .onResultOf(IJobInstanceUpdateEvent::getTimestampMs);
  private static final Function<IJobInstanceUpdateEvent, Integer> EVENT_TO_INSTANCE =
      IJobInstanceUpdateEvent::getInstanceId;

  private IJobUpdateDetails getDetails() {
    return storage.read(
        storeProvider -> storeProvider.getJobUpdateStore().fetchJobUpdate(UPDATE_ID).get());
  }

  private IJobUpdateDetails getDetails(IJobUpdateKey key) {
    return storage.read(
        storeProvider -> storeProvider.getJobUpdateStore().fetchJobUpdate(key).get());
  }

  private void assertLatestUpdateMessage(String expected) {
    IJobUpdateDetails details = getDetails();
    assertEquals(expected, Iterables.getLast(details.getUpdateEvents()).getMessage());
  }

  private void assertState(
      JobUpdateStatus expected,
      Multimap<Integer, JobUpdateAction> expectedActions) {

    assertStateUpdate(UPDATE_ID, expected, expectedActions);
  }

  private void assertStateUpdate(
      IJobUpdateKey key,
      JobUpdateStatus expected,
      Multimap<Integer, JobUpdateAction> expectedActions) {

    IJobUpdateDetails details = getDetails(key);
    Iterable<IJobInstanceUpdateEvent> orderedEvents =
        EVENT_ORDER.sortedCopy(details.getInstanceEvents());
    Multimap<Integer, IJobInstanceUpdateEvent> eventsByInstance =
        Multimaps.index(orderedEvents, EVENT_TO_INSTANCE);
    Multimap<Integer, JobUpdateAction> actionsByInstance =
        Multimaps.transformValues(eventsByInstance, JobUpdateControllerImpl.EVENT_TO_ACTION);
    assertEquals(expectedActions, actionsByInstance);
    assertEquals(expected, details.getUpdate().getSummary().getState().getStatus());
  }

  private IExpectationSetters<String> expectTaskKilled() {
    driver.killTask(EasyMock.anyObject());
    return expectLastCall();
  }

  private void insertPendingTasks(ITaskConfig task, Set<Integer> instanceIds) {
    storage.write((NoResult.Quiet) storeProvider ->
        stateManager.insertPendingTasks(storeProvider, task, instanceIds));
  }

  private void insertInitialTasks(IJobUpdate update) {
    storage.write((NoResult.Quiet) storeProvider -> {
      for (IInstanceTaskConfig config : update.getInstructions().getInitialState()) {
        insertPendingTasks(config.getTask(), expandInstanceIds(ImmutableSet.of(config)));
      }
    });
  }

  private void assertJobState(IJobKey job, Map<Integer, ITaskConfig> expected) {
    Iterable<IScheduledTask> tasks =
        Storage.Util.fetchTasks(storage, Query.jobScoped(job).active());

    Map<Integer, IScheduledTask> tasksByInstance =
        Maps.uniqueIndex(tasks, Tasks::getInstanceId);
    assertEquals(
        expected,
        ImmutableMap.copyOf(Maps.transformValues(tasksByInstance, Tasks::getConfig)));
  }

  @Test
  public void testSuccessfulUpdate() throws Exception {
    expectTaskKilled();

    control.replay();

    IJobUpdate update = makeJobUpdate(
        // No-op - task is already matching the new config.
        makeInstanceConfig(0, 0, NEW_CONFIG),
        // Task needing update.
        makeInstanceConfig(2, 2, OLD_CONFIG));
    insertInitialTasks(update);

    changeState(JOB, 0, ASSIGNED, STARTING, RUNNING);
    changeState(JOB, 2, ASSIGNED, STARTING, RUNNING);
    clock.advance(WATCH_TIMEOUT);

    ImmutableMultimap.Builder<Integer, JobUpdateAction> actions = ImmutableMultimap.builder();

    // Instance 1 is added
    updater.start(update, AUDIT);
    actions.putAll(1, INSTANCE_UPDATING);
    assertState(ROLLING_FORWARD, actions.build());
    changeState(JOB, 1, ASSIGNED, STARTING, RUNNING);

    // Updates may be paused for arbitrarily-long amounts of time, and the updater should not
    // take action while paused.
    updater.pause(UPDATE_ID, AUDIT);
    updater.pause(UPDATE_ID, AUDIT);  // Pausing again is a no-op.
    assertState(ROLL_FORWARD_PAUSED, actions.build());
    clock.advance(ONE_DAY);
    changeState(JOB, 1, FAILED, ASSIGNED, STARTING, RUNNING);
    changeState(JOB, 2, FAILED, ASSIGNED, STARTING, RUNNING);
    clock.advance(WATCH_TIMEOUT);
    updater.resume(UPDATE_ID, AUDIT);

    actions.putAll(1, INSTANCE_UPDATED).put(2, INSTANCE_UPDATING);
    assertState(ROLLING_FORWARD, actions.build());

    // A task outside the scope of the update should be ignored by the updater.
    insertPendingTasks(NEW_CONFIG, ImmutableSet.of(100));

    // Instance 2 is updated
    changeState(JOB, 2, KILLED, ASSIGNED, STARTING, RUNNING);
    clock.advance(WATCH_TIMEOUT);

    actions.put(2, INSTANCE_UPDATED);
    assertState(ROLLED_FORWARD, actions.build());
    assertJobState(
        JOB,
        ImmutableMap.of(0, NEW_CONFIG, 1, NEW_CONFIG, 2, NEW_CONFIG, 100, NEW_CONFIG));

    // Attempting to abort a finished update should fail.
    try {
      updater.abort(UPDATE_ID, AUDIT);
      fail("It should not be possible to abort a completed update.");
    } catch (UpdateStateException e) {
      // Expected.
    }
  }

  @Test
  public void testSuccessfulCoordinatedUpdate() throws Exception {
    expectTaskKilled().times(2);

    control.replay();

    JobUpdate builder = makeJobUpdate(
        // No-op - task is already matching the new config.
        makeInstanceConfig(0, 0, NEW_CONFIG),
        // Tasks needing update.
        makeInstanceConfig(1, 2, OLD_CONFIG)).newBuilder();

    builder.getInstructions().getSettings().setBlockIfNoPulsesAfterMs((int) PULSE_TIMEOUT_MS);
    insertInitialTasks(IJobUpdate.build(builder));

    changeState(JOB, 0, ASSIGNED, STARTING, RUNNING);
    changeState(JOB, 1, ASSIGNED, STARTING, RUNNING);
    changeState(JOB, 2, ASSIGNED, STARTING, RUNNING);
    clock.advance(WATCH_TIMEOUT);

    ImmutableMultimap.Builder<Integer, JobUpdateAction> actions = ImmutableMultimap.builder();
    updater.start(IJobUpdate.build(builder), AUDIT);

    // The update is blocked initially waiting for a pulse.
    assertState(ROLL_FORWARD_AWAITING_PULSE, actions.build());

    // Pulse arrives and update starts.
    assertEquals(JobUpdatePulseStatus.OK, updater.pulse(UPDATE_ID));
    changeState(JOB, 1, KILLED, ASSIGNED, STARTING, RUNNING);
    actions.put(1, INSTANCE_UPDATING);
    assertState(ROLLING_FORWARD, actions.build());
    clock.advance(WATCH_TIMEOUT);
    actions.put(1, INSTANCE_UPDATED);

    // The update is blocked due to expired pulse timeout.
    clock.advance(Amount.of(PULSE_TIMEOUT_MS, Time.MILLISECONDS));
    actions.put(2, INSTANCE_UPDATING);
    changeState(JOB, 2, KILLED);
    assertState(ROLL_FORWARD_AWAITING_PULSE, actions.build());
    assertLatestUpdateMessage(JobUpdateControllerImpl.PULSE_TIMEOUT_MESSAGE);

    // Pulse arrives and instance 2 is updated.
    assertEquals(JobUpdatePulseStatus.OK, updater.pulse(UPDATE_ID));
    changeState(JOB, 2, ASSIGNED, STARTING, RUNNING);
    clock.advance(WATCH_TIMEOUT);
    actions.put(2, INSTANCE_UPDATED);

    assertState(ROLLED_FORWARD, actions.build());
    assertJobState(JOB, ImmutableMap.of(0, NEW_CONFIG, 1, NEW_CONFIG, 2, NEW_CONFIG));
    assertEquals(JobUpdatePulseStatus.FINISHED, updater.pulse(UPDATE_ID));
  }

  @Test
  public void testRecoverCoordinatedUpdateFromStorage() throws Exception {
    expectTaskKilled().times(2);

    control.replay();

    JobUpdate builder =
        setInstanceCount(makeJobUpdate(makeInstanceConfig(0, 1, OLD_CONFIG)), 2).newBuilder();
    builder.getInstructions().getSettings().setBlockIfNoPulsesAfterMs((int) PULSE_TIMEOUT_MS);
    IJobUpdate update = IJobUpdate.build(builder);
    insertInitialTasks(update);
    changeState(JOB, 0, ASSIGNED, STARTING, RUNNING);
    changeState(JOB, 1, ASSIGNED, STARTING, RUNNING);
    clock.advance(ONE_DAY);

    storage.write((NoResult.Quiet) storeProvider ->
        saveJobUpdate(storeProvider.getJobUpdateStore(), update, ROLLING_FORWARD));

    clock.advance(ONE_MINUTE);

    subscriber.startAsync().awaitRunning();
    ImmutableMultimap.Builder<Integer, JobUpdateAction> actions = ImmutableMultimap.builder();

    // The update is blocked initially waiting for a pulse.
    assertState(ROLL_FORWARD_AWAITING_PULSE, actions.build());

    assertEquals(JobUpdatePulseStatus.OK, updater.pulse(UPDATE_ID));

    // Instance 0 is updated.
    changeState(JOB, 0, KILLED, ASSIGNED, STARTING, RUNNING);
    clock.advance(WATCH_TIMEOUT);

    // Instance 1 is updated.
    changeState(JOB, 1, KILLED, ASSIGNED, STARTING, RUNNING);
    clock.advance(WATCH_TIMEOUT);

    actions.putAll(0, INSTANCE_UPDATING, INSTANCE_UPDATED)
        .putAll(1, INSTANCE_UPDATING, INSTANCE_UPDATED);

    assertState(ROLLED_FORWARD, actions.build());
    assertEquals(JobUpdatePulseStatus.FINISHED, updater.pulse(UPDATE_ID));
  }

  @Test
  public void testRecoverLongPulseTimeoutCoordinatedUpdateFromStorage() throws Exception {
    // A brief failover in the middle of a rolling forward update with a long pulse timeout should
    // mean that after scheduler startup the update is not waiting for a pulse.
    expectTaskKilled().times(1);

    control.replay();

    JobUpdate builder =
        setInstanceCount(makeJobUpdate(makeInstanceConfig(0, 0, OLD_CONFIG)), 1).newBuilder();
    builder.getInstructions().getSettings()
        .setBlockIfNoPulsesAfterMs(Ints.checkedCast(ONE_HOUR.as(Time.MILLISECONDS)));
    IJobUpdate update = IJobUpdate.build(builder);
    insertInitialTasks(update);
    changeState(JOB, 0, ASSIGNED, STARTING, RUNNING);
    clock.advance(ONE_DAY);

    storage.write((NoResult.Quiet) storeProvider ->
        saveJobUpdate(storeProvider.getJobUpdateStore(), update, ROLL_FORWARD_AWAITING_PULSE));

    // The first pulse comes after one minute
    clock.advance(ONE_MINUTE);

    storage.write(
        (NoResult.Quiet) storeProvider ->
            saveJobUpdateEvent(storeProvider.getJobUpdateStore(), update, ROLLING_FORWARD));

    clock.advance(ONE_MINUTE);

    subscriber.startAsync().awaitRunning();
    ImmutableMultimap.Builder<Integer, JobUpdateAction> actions = ImmutableMultimap.builder();

    actions.putAll(0, INSTANCE_UPDATING);
    // Since the pulse interval is so large and the downtime was so short, the update does not need
    // to wait for a pulse.
    assertState(ROLLING_FORWARD, actions.build());

    // Instance 0 is updated.
    changeState(JOB, 0, KILLED, ASSIGNED, STARTING, RUNNING);
    clock.advance(WATCH_TIMEOUT);

    actions.putAll(0, INSTANCE_UPDATED);

    assertState(ROLLED_FORWARD, actions.build());
    assertEquals(JobUpdatePulseStatus.FINISHED, updater.pulse(UPDATE_ID));
  }

  @Test
  public void testRecoverAwaitingPulseFromStorage() throws Exception {
    expectTaskKilled();

    control.replay();

    JobUpdate builder =
        setInstanceCount(makeJobUpdate(makeInstanceConfig(0, 0, OLD_CONFIG)), 1).newBuilder();
    builder.getInstructions().getSettings().setBlockIfNoPulsesAfterMs((int) PULSE_TIMEOUT_MS);
    IJobUpdate update = IJobUpdate.build(builder);
    insertInitialTasks(update);
    changeState(JOB, 0, ASSIGNED, STARTING, RUNNING);
    clock.advance(ONE_DAY);

    storage.write((NoResult.Quiet) storeProvider ->
        saveJobUpdate(storeProvider.getJobUpdateStore(), update, ROLL_FORWARD_AWAITING_PULSE));

    subscriber.startAsync().awaitRunning();
    ImmutableMultimap.Builder<Integer, JobUpdateAction> actions = ImmutableMultimap.builder();

    assertState(ROLL_FORWARD_AWAITING_PULSE, actions.build());
    assertEquals(JobUpdatePulseStatus.OK, updater.pulse(UPDATE_ID));

    changeState(JOB, 0, KILLED, ASSIGNED, STARTING, RUNNING);
    clock.advance(WATCH_TIMEOUT);
    actions.putAll(0, INSTANCE_UPDATING, INSTANCE_UPDATED);

    assertState(ROLLED_FORWARD, actions.build());
    assertEquals(JobUpdatePulseStatus.FINISHED, updater.pulse(UPDATE_ID));
  }

  @Test
  public void testRecoverCoordinatedPausedFromStorage() throws Exception {
    expectTaskKilled();

    control.replay();

    JobUpdate builder =
        setInstanceCount(makeJobUpdate(makeInstanceConfig(0, 0, OLD_CONFIG)), 1).newBuilder();
    builder.getInstructions().getSettings().setBlockIfNoPulsesAfterMs((int) PULSE_TIMEOUT_MS);
    IJobUpdate update = IJobUpdate.build(builder);
    insertInitialTasks(update);
    changeState(JOB, 0, ASSIGNED, STARTING, RUNNING);
    clock.advance(ONE_DAY);

    storage.write((NoResult.Quiet) storeProvider ->
        saveJobUpdate(storeProvider.getJobUpdateStore(), update, ROLL_FORWARD_PAUSED));

    subscriber.startAsync().awaitRunning();
    ImmutableMultimap.Builder<Integer, JobUpdateAction> actions = ImmutableMultimap.builder();

    assertState(ROLL_FORWARD_PAUSED, actions.build());
    assertEquals(JobUpdatePulseStatus.OK, updater.pulse(UPDATE_ID));

    updater.resume(UPDATE_ID, AUDIT);

    changeState(JOB, 0, KILLED, ASSIGNED, STARTING, RUNNING);
    clock.advance(WATCH_TIMEOUT);
    actions.putAll(0, INSTANCE_UPDATING, INSTANCE_UPDATED);

    assertState(ROLLED_FORWARD, actions.build());
    assertEquals(JobUpdatePulseStatus.FINISHED, updater.pulse(UPDATE_ID));
  }

  @Test
  public void testResumeToAwaitingPulse() throws Exception {
    expectTaskKilled().times(2);

    control.replay();

    JobUpdate builder =
        setInstanceCount(makeJobUpdate(makeInstanceConfig(0, 1, OLD_CONFIG)), 2).newBuilder();
    builder.getInstructions().getSettings().setBlockIfNoPulsesAfterMs((int) PULSE_TIMEOUT_MS);
    IJobUpdate update = IJobUpdate.build(builder);
    insertInitialTasks(update);
    changeState(JOB, 0, ASSIGNED, STARTING, RUNNING);
    changeState(JOB, 1, ASSIGNED, STARTING, RUNNING);
    clock.advance(ONE_DAY);

    ImmutableMultimap.Builder<Integer, JobUpdateAction> actions = ImmutableMultimap.builder();
    updater.start(IJobUpdate.build(builder), AUDIT);

    // The update is blocked initially waiting for a pulse.
    assertState(ROLL_FORWARD_AWAITING_PULSE, actions.build());

    // Pause the awaiting pulse update.
    updater.pause(UPDATE_ID, AUDIT);
    assertState(ROLL_FORWARD_PAUSED, actions.build());

    // Resume into awaiting pulse state.
    updater.resume(UPDATE_ID, AUDIT);
    assertState(ROLL_FORWARD_AWAITING_PULSE, actions.build());

    assertEquals(JobUpdatePulseStatus.OK, updater.pulse(UPDATE_ID));

    // Instance 0 is updated.
    changeState(JOB, 0, KILLED, ASSIGNED, STARTING, RUNNING);
    clock.advance(WATCH_TIMEOUT);

    // Instance 1 is updated.
    changeState(JOB, 1, KILLED, ASSIGNED, STARTING, RUNNING);
    clock.advance(WATCH_TIMEOUT);

    actions.putAll(0, INSTANCE_UPDATING, INSTANCE_UPDATED)
        .putAll(1, INSTANCE_UPDATING, INSTANCE_UPDATED);

    assertState(ROLLED_FORWARD, actions.build());
    assertEquals(JobUpdatePulseStatus.FINISHED, updater.pulse(UPDATE_ID));
  }

  @Test
  public void testPulsePausedUpdate() throws Exception {
    expectTaskKilled().times(2);

    control.replay();

    JobUpdate builder = makeJobUpdate(
        // No-op - task is already matching the new config.
        makeInstanceConfig(0, 0, NEW_CONFIG),
        // Tasks needing update.
        makeInstanceConfig(1, 2, OLD_CONFIG)).newBuilder();

    builder.getInstructions().getSettings().setBlockIfNoPulsesAfterMs((int) PULSE_TIMEOUT_MS);
    insertInitialTasks(IJobUpdate.build(builder));

    changeState(JOB, 0, ASSIGNED, STARTING, RUNNING);
    changeState(JOB, 1, ASSIGNED, STARTING, RUNNING);
    changeState(JOB, 2, ASSIGNED, STARTING, RUNNING);
    clock.advance(WATCH_TIMEOUT);

    ImmutableMultimap.Builder<Integer, JobUpdateAction> actions = ImmutableMultimap.builder();
    updater.start(IJobUpdate.build(builder), AUDIT);

    // The update is blocked initially waiting for a pulse.
    assertState(ROLL_FORWARD_AWAITING_PULSE, actions.build());

    // Pulse arrives and update starts.
    assertEquals(JobUpdatePulseStatus.OK, updater.pulse(UPDATE_ID));
    changeState(JOB, 1, KILLED, ASSIGNED, STARTING, RUNNING);
    actions.put(1, INSTANCE_UPDATING);
    clock.advance(WATCH_TIMEOUT);
    actions.put(1, INSTANCE_UPDATED);
    actions.put(2, INSTANCE_UPDATING);
    clock.advance(Amount.of(PULSE_TIMEOUT_MS, Time.MILLISECONDS));

    // Update is paused
    updater.pause(UPDATE_ID, AUDIT);
    assertState(ROLL_FORWARD_PAUSED, actions.build());

    // A paused update is pulsed.
    assertEquals(JobUpdatePulseStatus.OK, updater.pulse(UPDATE_ID));

    // Update is resumed
    updater.resume(UPDATE_ID, AUDIT);
    assertState(ROLLING_FORWARD, actions.build());

    // Instance 2 is updated.
    changeState(JOB, 2, KILLED, ASSIGNED, STARTING, RUNNING);
    clock.advance(WATCH_TIMEOUT);
    actions.put(2, INSTANCE_UPDATED);

    assertState(ROLLED_FORWARD, actions.build());
    assertJobState(JOB, ImmutableMap.of(0, NEW_CONFIG, 1, NEW_CONFIG, 2, NEW_CONFIG));

    assertEquals(JobUpdatePulseStatus.FINISHED, updater.pulse(UPDATE_ID));
  }

  @Test
  public void testUnblockDeletedUpdate() throws Exception {
    control.replay();

    JobUpdate builder =
        setInstanceCount(makeJobUpdate(makeInstanceConfig(0, 1, OLD_CONFIG)), 2).newBuilder();
    builder.getInstructions().getSettings().setBlockIfNoPulsesAfterMs((int) PULSE_TIMEOUT_MS);
    IJobUpdate update = IJobUpdate.build(builder);
    insertInitialTasks(update);
    changeState(JOB, 0, ASSIGNED, STARTING, RUNNING);
    changeState(JOB, 1, ASSIGNED, STARTING, RUNNING);
    clock.advance(ONE_DAY);

    storage.write((NoResult.Quiet) storeProvider ->
        saveJobUpdate(storeProvider.getJobUpdateStore(), update, ROLLING_FORWARD));

    clock.advance(ONE_MINUTE);

    subscriber.startAsync().awaitRunning();

    ImmutableMultimap.Builder<Integer, JobUpdateAction> actions = ImmutableMultimap.builder();
    // The update is blocked initially waiting for a pulse.
    assertState(ROLL_FORWARD_AWAITING_PULSE, actions.build());

    storage.write((NoResult.Quiet) storeProvider -> {
      storeProvider.getJobUpdateStore().deleteAllUpdates();
    });

    // The pulse still returns OK but the error is handled.
    assertEquals(JobUpdatePulseStatus.OK, updater.pulse(UPDATE_ID));
  }

  @Test
  public void testPulseInvalidUpdateId() throws Exception {
    control.replay();

    assertEquals(
        JobUpdatePulseStatus.FINISHED,
        updater.pulse(IJobUpdateKey.build(new JobUpdateKey(JOB.newBuilder(), "invalid"))));
  }

  @Test(expected = IllegalStateException.class)
  public void testShutdownOnFailedPulse() throws Exception {
    // Missing kill expectation will trigger failure.
    shutdownCommand.execute();
    expectLastCall().andAnswer(() -> {
      throw new IllegalStateException("Expected shutdown triggered.");
    });

    control.replay();

    JobUpdate builder = makeJobUpdate(
        // No-op - task is already matching the new config.
        makeInstanceConfig(0, 0, NEW_CONFIG),
        // Tasks needing update.
        makeInstanceConfig(1, 2, OLD_CONFIG)).newBuilder();

    builder.getInstructions().getSettings().setBlockIfNoPulsesAfterMs((int) PULSE_TIMEOUT_MS);
    insertInitialTasks(IJobUpdate.build(builder));

    changeState(JOB, 0, ASSIGNED, STARTING, RUNNING);
    changeState(JOB, 1, ASSIGNED, STARTING, RUNNING);
    changeState(JOB, 2, ASSIGNED, STARTING, RUNNING);
    clock.advance(WATCH_TIMEOUT);

    ImmutableMultimap.Builder<Integer, JobUpdateAction> actions = ImmutableMultimap.builder();
    updater.start(IJobUpdate.build(builder), AUDIT);

    // The update is blocked initially waiting for a pulse.
    assertState(ROLL_FORWARD_AWAITING_PULSE, actions.build());

    // Pulse arrives and update starts.
    assertEquals(JobUpdatePulseStatus.OK, updater.pulse(UPDATE_ID));
    changeState(JOB, 1, KILLED, ASSIGNED, STARTING, RUNNING);
  }

  @Test
  public void testSuccessfulBatchedUpdate() throws Exception {
    expectTaskKilled().times(3);

    control.replay();

    JobUpdate builder = makeJobUpdate(makeInstanceConfig(0, 2, OLD_CONFIG)).newBuilder();
    builder.getInstructions().getSettings()
        .setUpdateStrategy(
            JobUpdateStrategy.batchStrategy(new BatchJobUpdateStrategy().setGroupSize(2)));
    IJobUpdate update = IJobUpdate.build(builder);
    insertInitialTasks(update);

    changeState(JOB, 0, ASSIGNED, STARTING, RUNNING);
    changeState(JOB, 1, ASSIGNED, STARTING, RUNNING);
    changeState(JOB, 2, ASSIGNED, STARTING, RUNNING);
    clock.advance(WATCH_TIMEOUT);

    ImmutableMultimap.Builder<Integer, JobUpdateAction> actions = ImmutableMultimap.builder();

    // Instances 0 and 1 are updated.
    updater.start(update, AUDIT);
    actions.putAll(0, INSTANCE_UPDATING)
        .putAll(1, INSTANCE_UPDATING);
    assertState(ROLLING_FORWARD, actions.build());
    changeState(JOB, 1, FINISHED, ASSIGNED, STARTING, RUNNING);
    clock.advance(Amount.of(WATCH_TIMEOUT.getValue() / 2, Time.MILLISECONDS));
    changeState(JOB, 0, FINISHED, ASSIGNED, STARTING, RUNNING);
    clock.advance(Amount.of(WATCH_TIMEOUT.getValue() / 2, Time.MILLISECONDS));

    // Instance 1 finished first, but update does not yet proceed until 0 finishes.
    actions.putAll(1, INSTANCE_UPDATED);
    assertState(ROLLING_FORWARD, actions.build());
    clock.advance(WATCH_TIMEOUT);
    actions.putAll(0, INSTANCE_UPDATED);

    // Instance 2 is updated.
    changeState(JOB, 2, FINISHED, ASSIGNED, STARTING, RUNNING);
    clock.advance(WATCH_TIMEOUT);
    actions.putAll(2, INSTANCE_UPDATING, INSTANCE_UPDATED);
    assertState(ROLLED_FORWARD, actions.build());

    assertJobState(
        JOB,
        ImmutableMap.of(0, NEW_CONFIG, 1, NEW_CONFIG, 2, NEW_CONFIG));
  }

  @Test
  public void testUpdateSpecificInstances() throws Exception {
    expectTaskKilled();

    control.replay();

    JobUpdate builder =
        setInstanceCount(makeJobUpdate(makeInstanceConfig(0, 0, OLD_CONFIG)), 1).newBuilder();
    builder.getInstructions().getSettings().setUpdateOnlyTheseInstances(
        ImmutableSet.of(new Range(0, 0)));
    IJobUpdate update = IJobUpdate.build(builder);
    insertPendingTasks(OLD_CONFIG, ImmutableSet.of(0, 1));

    changeState(JOB, 0, ASSIGNED, STARTING, RUNNING);
    changeState(JOB, 1, ASSIGNED, STARTING, RUNNING);
    clock.advance(WATCH_TIMEOUT);

    // Instance 0 is updated
    updater.start(update, AUDIT);
    changeState(JOB, 0, KILLED, ASSIGNED, STARTING, RUNNING);
    clock.advance(WATCH_TIMEOUT);

    ImmutableMultimap.Builder<Integer, JobUpdateAction> actions = ImmutableMultimap.builder();
    assertState(
        ROLLED_FORWARD,
        actions.putAll(0, INSTANCE_UPDATING, INSTANCE_UPDATED).build());
    assertJobState(
        JOB,
        ImmutableMap.of(0, NEW_CONFIG, 1, OLD_CONFIG));
  }

  @Test
  public void testUpdateSpecificInstancesSkipUnchanged() throws Exception {
    control.replay();

    JobUpdate builder = makeJobUpdate().newBuilder();

    builder.getInstructions().getDesiredState().setInstances(ImmutableSet.of(new Range(1, 1)));
    builder.getInstructions().getSettings().setUpdateOnlyTheseInstances(
        ImmutableSet.of(new Range(0, 1)));
    IJobUpdate update = IJobUpdate.build(builder);
    insertPendingTasks(NEW_CONFIG, ImmutableSet.of(0));
    insertPendingTasks(OLD_CONFIG, ImmutableSet.of(2));

    changeState(JOB, 0, ASSIGNED, STARTING, RUNNING);
    changeState(JOB, 2, ASSIGNED, STARTING, RUNNING);
    clock.advance(WATCH_TIMEOUT);

    // Instance 1 is added, while instance 0 is skipped
    updater.start(update, AUDIT);
    changeState(JOB, 1, ASSIGNED, STARTING, RUNNING);
    clock.advance(WATCH_TIMEOUT);

    ImmutableMultimap.Builder<Integer, JobUpdateAction> actions = ImmutableMultimap.builder();
    assertState(
        ROLLED_FORWARD,
        actions.putAll(1, INSTANCE_UPDATING, INSTANCE_UPDATED).build());
    assertJobState(
        JOB,
        ImmutableMap.of(0, NEW_CONFIG, 1, NEW_CONFIG, 2, OLD_CONFIG));
  }

  @Test
  public void testRollback() throws Exception {
    expectTaskKilled().times(4);

    control.replay();

    IJobUpdate update = makeJobUpdate(
            makeInstanceConfig(0, 0, OLD_CONFIG),
            makeInstanceConfig(2, 3, OLD_CONFIG));
    insertInitialTasks(update);

    changeState(JOB, 0, ASSIGNED, STARTING, RUNNING);
    changeState(JOB, 2, ASSIGNED, STARTING, RUNNING);
    changeState(JOB, 3, ASSIGNED, STARTING, RUNNING);
    clock.advance(WATCH_TIMEOUT);

    ImmutableMultimap.Builder<Integer, JobUpdateAction> actions = ImmutableMultimap.builder();

    // Instance 1 is added.
    updater.start(update, AUDIT);
    actions.putAll(1, INSTANCE_UPDATING);
    assertState(ROLLING_FORWARD, actions.build());
    changeState(JOB, 1, ASSIGNED, STARTING, RUNNING);
    clock.advance(WATCH_TIMEOUT);

    // Instance 0 is updated.
    changeState(JOB, 0, KILLED, ASSIGNED, STARTING, RUNNING);
    actions.putAll(1, INSTANCE_UPDATED)
        .putAll(0, INSTANCE_UPDATING, INSTANCE_UPDATED);
    clock.advance(WATCH_TIMEOUT);

    // Instance 2 is updated, but fails.
    changeState(JOB, 2, KILLED, ASSIGNED, STARTING, RUNNING);
    actions.putAll(2, INSTANCE_UPDATING, INSTANCE_UPDATE_FAILED, INSTANCE_ROLLING_BACK);
    clock.advance(FLAPPING_THRESHOLD);
    changeState(JOB, 2, FAILED);

    // Instance 2 is rolled back.
    assertState(ROLLING_BACK, actions.build());
    assertLatestUpdateMessage(JobUpdateControllerImpl.failureMessage(2, Failure.EXITED));
    changeState(JOB, 2, ASSIGNED, STARTING, RUNNING);
    actions.putAll(0, INSTANCE_ROLLING_BACK)
        .putAll(2, INSTANCE_ROLLED_BACK);
    clock.advance(WATCH_TIMEOUT);

    // A rollback may be paused.
    updater.pause(UPDATE_ID, AUDIT);
    assertState(ROLL_BACK_PAUSED, actions.build());
    clock.advance(ONE_DAY);
    updater.resume(UPDATE_ID, AUDIT);
    assertState(ROLLING_BACK, actions.build());

    // Instance 0 is rolled back.
    changeState(JOB, 0, KILLED, ASSIGNED, STARTING, RUNNING);
    actions.putAll(0, INSTANCE_ROLLED_BACK);
    clock.advance(WATCH_TIMEOUT);

    // Instance 1 is removed.
    changeState(JOB, 1, KILLED);
    actions.putAll(1, INSTANCE_ROLLING_BACK, INSTANCE_ROLLED_BACK);
    clock.advance(WATCH_TIMEOUT);

    assertState(ROLLED_BACK, actions.build());
    assertJobState(JOB, ImmutableMap.of(0, OLD_CONFIG, 2, OLD_CONFIG, 3, OLD_CONFIG));
  }

  @Test
  public void testRollbackDisabled() throws Exception {
    expectTaskKilled().times(2);

    control.replay();

    JobUpdate builder = makeJobUpdate(
        makeInstanceConfig(0, 0, OLD_CONFIG),
        makeInstanceConfig(2, 3, OLD_CONFIG))
        .newBuilder();
    builder.getInstructions().getSettings().setRollbackOnFailure(false);
    IJobUpdate update = IJobUpdate.build(builder);
    insertInitialTasks(update);

    changeState(JOB, 0, ASSIGNED, STARTING, RUNNING);
    changeState(JOB, 2, ASSIGNED, STARTING, RUNNING);
    changeState(JOB, 3, ASSIGNED, STARTING, RUNNING);
    clock.advance(WATCH_TIMEOUT);

    ImmutableMultimap.Builder<Integer, JobUpdateAction> actions = ImmutableMultimap.builder();

    // Instance 1 is added.
    updater.start(update, AUDIT);
    actions.putAll(1, INSTANCE_UPDATING);
    assertState(ROLLING_FORWARD, actions.build());
    changeState(JOB, 1, ASSIGNED, STARTING, RUNNING);
    clock.advance(WATCH_TIMEOUT);

    // Instance 0 is updated.
    changeState(JOB, 0, KILLED, ASSIGNED, STARTING, RUNNING);
    actions.putAll(1, INSTANCE_UPDATED)
        .putAll(0, INSTANCE_UPDATING, INSTANCE_UPDATED);
    clock.advance(WATCH_TIMEOUT);

    // Instance 2 is updated, but fails.
    changeState(JOB, 2, KILLED, ASSIGNED, STARTING, RUNNING);
    actions.putAll(2, INSTANCE_UPDATING, INSTANCE_UPDATE_FAILED);
    clock.advance(FLAPPING_THRESHOLD);
    changeState(JOB, 2, FAILED);
    clock.advance(WATCH_TIMEOUT);

    // Rollback is disabled, update fails.
    assertState(JobUpdateStatus.FAILED, actions.build());
  }

  @Test
  public void testAbort() throws Exception {
    expectTaskKilled();

    control.replay();

    IJobUpdate update = makeJobUpdate(makeInstanceConfig(0, 2, OLD_CONFIG));
    insertInitialTasks(update);

    changeState(JOB, 0, ASSIGNED, STARTING, RUNNING);
    changeState(JOB, 2, ASSIGNED, STARTING, RUNNING);
    clock.advance(WATCH_TIMEOUT);

    // Instance 0 is updated
    updater.start(update, AUDIT);
    changeState(JOB, 0, KILLED, ASSIGNED, STARTING, RUNNING);
    clock.advance(WATCH_TIMEOUT);

    ImmutableMultimap.Builder<Integer, JobUpdateAction> actions = ImmutableMultimap.builder();
    actions.putAll(0, INSTANCE_UPDATING, INSTANCE_UPDATED)
        .putAll(1, INSTANCE_UPDATING);

    updater.abort(UPDATE_ID, AUDIT);
    assertState(ABORTED, actions.build());
    clock.advance(WATCH_TIMEOUT);
    assertJobState(JOB, ImmutableMap.of(0, NEW_CONFIG, 1, NEW_CONFIG, 2, OLD_CONFIG));
  }

  @Test
  public void testRollbackFailed() throws Exception {
    expectTaskKilled().times(2);

    control.replay();

    IJobUpdate update = makeJobUpdate(
        makeInstanceConfig(0, 2, OLD_CONFIG));
    insertInitialTasks(update);

    changeState(JOB, 0, ASSIGNED, STARTING, RUNNING);
    changeState(JOB, 1, ASSIGNED, STARTING, RUNNING);
    changeState(JOB, 2, ASSIGNED, STARTING, RUNNING);
    clock.advance(WATCH_TIMEOUT);

    ImmutableMultimap.Builder<Integer, JobUpdateAction> actions = ImmutableMultimap.builder();

    // Instance 0 is updated.
    updater.start(update, AUDIT);
    actions.putAll(0, INSTANCE_UPDATING);
    assertState(ROLLING_FORWARD, actions.build());
    changeState(JOB, 0, KILLED, ASSIGNED, STARTING, RUNNING);
    clock.advance(WATCH_TIMEOUT);

    // Instance 1 is updated, but fails.
    changeState(JOB, 1, KILLED, ASSIGNED, STARTING, RUNNING);
    clock.advance(FLAPPING_THRESHOLD);
    changeState(JOB, 1, FAILED);

    // Instance 1 is rolled back, but fails.
    actions.putAll(0, INSTANCE_UPDATED)
        .putAll(1, INSTANCE_UPDATING, INSTANCE_UPDATE_FAILED, INSTANCE_ROLLING_BACK);
    assertState(ROLLING_BACK, actions.build());
    changeState(JOB, 1, ASSIGNED, STARTING, RUNNING);
    clock.advance(FLAPPING_THRESHOLD);
    changeState(JOB, 1, FAILED);

    actions.putAll(1, INSTANCE_ROLLBACK_FAILED);
    assertState(JobUpdateStatus.FAILED, actions.build());
    clock.advance(WATCH_TIMEOUT);
    assertJobState(JOB, ImmutableMap.of(0, NEW_CONFIG, 1, OLD_CONFIG, 2, OLD_CONFIG));
  }

  private void expectInvalid(JobUpdate update)
      throws UpdateStateException, UpdateConfigurationException {

    try {
      updater.start(IJobUpdate.build(update), AUDIT);
      fail();
    } catch (IllegalArgumentException e) {
      // Expected.
    }
  }

  @Test
  public void testStartInvalidUpdate() throws Exception {
    control.replay();

    JobUpdate update = makeJobUpdate().newBuilder();
    update.getInstructions()
        .getSettings()
        .setUpdateStrategy(
            JobUpdateStrategy.queueStrategy(new QueueJobUpdateStrategy().setGroupSize(-1)));
    expectInvalid(update);

    update = makeJobUpdate().newBuilder();
    update.getInstructions().getSettings().setMinWaitInInstanceRunningMs(-1);
    expectInvalid(update);
  }

  @Test
  public void testConfigurationPolicyChange() throws Exception {
    // Simulates a change in input validation after a job update has been persisted.

    expectTaskKilled().times(2);

    control.replay();

    IJobUpdate update = setInstanceCount(
        makeJobUpdate(makeInstanceConfig(0, 1, OLD_CONFIG)), 2);
    insertInitialTasks(update);

    changeState(JOB, 0, ASSIGNED, STARTING, RUNNING);
    changeState(JOB, 1, ASSIGNED, STARTING, RUNNING);
    clock.advance(WATCH_TIMEOUT);

    ImmutableMultimap.Builder<Integer, JobUpdateAction> actions = ImmutableMultimap.builder();

    // Instance 0 is updated
    updater.start(update, AUDIT);
    actions.putAll(0, INSTANCE_UPDATING);
    assertState(ROLLING_FORWARD, actions.build());

    storage.write((NoResult.Quiet) storeProvider -> {
      JobUpdateStore.Mutable store = storeProvider.getJobUpdateStore();
      store.deleteAllUpdates();

      JobUpdate builder = update.newBuilder();
      builder.getInstructions()
          .getSettings()
          .getUpdateStrategy()
          .getQueueStrategy()
          .setGroupSize(0);
      saveJobUpdate(store, IJobUpdate.build(builder), ROLLING_FORWARD);
    });

    changeState(JOB, 0, KILLED, ASSIGNED, STARTING, RUNNING);
    clock.advance(WATCH_TIMEOUT);

    // Instance 1 is updated, but fails.
    changeState(JOB, 1, KILLED, ASSIGNED, STARTING, RUNNING, FAILED);
    // Actions is reset here since we wiped the updates tables earlier in the test case.
    actions = ImmutableMultimap.builder();
    actions.putAll(0, INSTANCE_UPDATED)
        .putAll(1, INSTANCE_UPDATING, INSTANCE_UPDATE_FAILED);
    clock.advance(WATCH_TIMEOUT);

    assertState(ERROR, actions.build());
  }

  private void saveJobUpdate(
      JobUpdateStore.Mutable store,
      IJobUpdate update,
      JobUpdateStatus status) {

    store.saveJobUpdate(update);
    saveJobUpdateEvent(store, update, status);
  }

  private void saveJobUpdateEvent(
      JobUpdateStore.Mutable store,
      IJobUpdate update,
      JobUpdateStatus status) {

    store.saveJobUpdateEvent(
        update.getSummary().getKey(),
        IJobUpdateEvent.build(
            new JobUpdateEvent()
                .setStatus(status)
                .setTimestampMs(clock.nowMillis())));
  }

  @Test
  public void testRecoverFromStorage() throws Exception {
    expectTaskKilled().times(2);

    control.replay();

    IJobUpdate update = setInstanceCount(makeJobUpdate(makeInstanceConfig(0, 1, OLD_CONFIG)), 2);
    insertInitialTasks(update);
    changeState(JOB, 0, ASSIGNED, STARTING, RUNNING);
    changeState(JOB, 1, ASSIGNED, STARTING, RUNNING);
    clock.advance(ONE_DAY);

    storage.write((NoResult.Quiet) storeProvider ->
        saveJobUpdate(storeProvider.getJobUpdateStore(), update, ROLLING_FORWARD));

    subscriber.startAsync().awaitRunning();

    // Instance 0 is updated.
    changeState(JOB, 0, KILLED, ASSIGNED, STARTING, RUNNING);
    clock.advance(WATCH_TIMEOUT);

    // Instance 1 is updated.
    changeState(JOB, 1, KILLED, ASSIGNED, STARTING, RUNNING);
    clock.advance(WATCH_TIMEOUT);

    ImmutableMultimap.Builder<Integer, JobUpdateAction> actions = ImmutableMultimap.builder();
    actions.putAll(0, INSTANCE_UPDATING, INSTANCE_UPDATED)
        .putAll(1, INSTANCE_UPDATING, INSTANCE_UPDATED);

    assertState(ROLLED_FORWARD, actions.build());
  }

  @Test
  public void testImmediatelySuccessfulUpdate() throws Exception {
    control.replay();

    IJobUpdate update = makeJobUpdate(makeInstanceConfig(0, 2, NEW_CONFIG));
    insertInitialTasks(update);
    changeState(JOB, 0, ASSIGNED, STARTING, RUNNING);
    changeState(JOB, 1, ASSIGNED, STARTING, RUNNING);
    changeState(JOB, 2, ASSIGNED, STARTING, RUNNING);
    clock.advance(ONE_DAY);
    updater.start(update, AUDIT);
  }

  @Test(expected = IllegalArgumentException.class)
  public void testNoopUpdateEmptyDiff() throws Exception {
    control.replay();

    IJobUpdate update = makeJobUpdate();
    JobUpdate builder = update.newBuilder();
    builder.getInstructions().unsetDesiredState();

    updater.start(IJobUpdate.build(builder), AUDIT);
  }

  @Test
  public void testSlowToScheduleTask() throws Exception {
    expectTaskKilled().times(2);

    control.replay();

    IJobUpdate update = setInstanceCount(makeJobUpdate(makeInstanceConfig(0, 1, OLD_CONFIG)), 2);
    insertInitialTasks(update);
    changeState(JOB, 0, ASSIGNED, STARTING, RUNNING);
    changeState(JOB, 1, ASSIGNED, STARTING, RUNNING);

    ImmutableMultimap.Builder<Integer, JobUpdateAction> actions = ImmutableMultimap.builder();

    // Instance 0 is updated.
    updater.start(update, AUDIT);
    actions.putAll(0, INSTANCE_UPDATING);
    assertState(ROLLING_FORWARD, actions.build());
    changeState(JOB, 0, KILLED, ASSIGNED, STARTING, RUNNING);
    clock.advance(WATCH_TIMEOUT);

    // Instance 1 is not advancing past PENDING.
    changeState(JOB, 1, KILLED);
    actions.putAll(0, INSTANCE_UPDATED)
        .putAll(1, INSTANCE_UPDATING);
    assertState(ROLLING_FORWARD, actions.build());

    updater.abort(update.getSummary().getKey(), AUDIT);
    assertState(ABORTED, actions.build());
  }

  @Test
  public void testAddInstances() throws Exception {
    control.replay();

    IJobUpdate update = makeJobUpdate();
    insertPendingTasks(NEW_CONFIG, ImmutableSet.of(0, 1));

    changeState(JOB, 0, ASSIGNED, STARTING, RUNNING);
    changeState(JOB, 1, ASSIGNED, STARTING, RUNNING);
    clock.advance(WATCH_TIMEOUT);

    ImmutableMultimap.Builder<Integer, JobUpdateAction> actions = ImmutableMultimap.builder();

    // Instance 2 is added
    updater.start(update, AUDIT);
    actions.putAll(2, INSTANCE_UPDATING);
    assertState(ROLLING_FORWARD, actions.build());
    changeState(JOB, 2, ASSIGNED, STARTING, RUNNING);

    clock.advance(WATCH_TIMEOUT);
    actions.putAll(2, INSTANCE_UPDATED);
    assertState(ROLLED_FORWARD, actions.build());

    assertJobState(
        JOB,
        ImmutableMap.of(0, NEW_CONFIG, 1, NEW_CONFIG, 2, NEW_CONFIG));
  }

  @Test
  public void testRemoveInstances() throws Exception {
    expectTaskKilled();

    control.replay();

    // Set instance count such that instance 1 is removed.
    IJobUpdate update = setInstanceCount(makeJobUpdate(makeInstanceConfig(0, 1, NEW_CONFIG)), 1);
    insertInitialTasks(update);

    changeState(JOB, 0, ASSIGNED, STARTING, RUNNING);
    changeState(JOB, 1, ASSIGNED, STARTING, RUNNING);
    clock.advance(WATCH_TIMEOUT);

    ImmutableMultimap.Builder<Integer, JobUpdateAction> actions = ImmutableMultimap.builder();

    // Instance 1 is removed.
    updater.start(update, AUDIT);
    actions.putAll(1, INSTANCE_UPDATING);
    changeState(JOB, 1, KILLED);
    clock.advance(WATCH_TIMEOUT);

    actions.put(1, INSTANCE_UPDATED);
    assertState(ROLLED_FORWARD, actions.build());
    assertJobState(JOB, ImmutableMap.of(0, NEW_CONFIG));
  }

  @Test
  public void testBadPubsubUpdate() {
    control.replay();

    subscriber.taskChangedState(
        PubsubEvent.TaskStateChange.transition(IScheduledTask.build(new ScheduledTask()), RUNNING));
  }

  @Test(expected = UpdateStateException.class)
  public void testPauseUnknownUpdate() throws Exception {
    control.replay();

    updater.pause(UPDATE_ID, AUDIT);
  }

  @Test(expected = UpdateStateException.class)
  public void testResumeUnknownUpdate() throws Exception {
    control.replay();

    updater.resume(UPDATE_ID, AUDIT);
  }

  @Test
  public void testFailToRollbackCompletedUpdate() throws Exception {
    expectTaskKilled().times(3);

    control.replay();

    JobUpdate builder = makeJobUpdate(makeInstanceConfig(0, 2, OLD_CONFIG)).newBuilder();
    builder.getInstructions().getSettings()
        .setUpdateStrategy(
            JobUpdateStrategy.batchStrategy(new BatchJobUpdateStrategy().setGroupSize(2)));
    IJobUpdate update = IJobUpdate.build(builder);
    insertInitialTasks(update);

    changeState(JOB, 0, ASSIGNED, STARTING, RUNNING);
    changeState(JOB, 1, ASSIGNED, STARTING, RUNNING);
    changeState(JOB, 2, ASSIGNED, STARTING, RUNNING);
    clock.advance(WATCH_TIMEOUT);

    ImmutableMultimap.Builder<Integer, JobUpdateAction> actions = ImmutableMultimap.builder();

    // Instances 0 and 1 are updated.
    updater.start(update, AUDIT);
    actions.putAll(0, INSTANCE_UPDATING)
        .putAll(1, INSTANCE_UPDATING);
    assertState(ROLLING_FORWARD, actions.build());
    changeState(JOB, 1, FINISHED, ASSIGNED, STARTING, RUNNING);
    clock.advance(Amount.of(WATCH_TIMEOUT.getValue() / 2, Time.MILLISECONDS));
    changeState(JOB, 0, FINISHED, ASSIGNED, STARTING, RUNNING);
    clock.advance(Amount.of(WATCH_TIMEOUT.getValue() / 2, Time.MILLISECONDS));

    // Instance 1 finished first, but update does not yet proceed until 0 finishes.
    actions.putAll(1, INSTANCE_UPDATED);
    assertState(ROLLING_FORWARD, actions.build());
    clock.advance(WATCH_TIMEOUT);
    actions.putAll(0, INSTANCE_UPDATED);

    // Instance 2 is updated.
    changeState(JOB, 2, FINISHED, ASSIGNED, STARTING, RUNNING);
    clock.advance(WATCH_TIMEOUT);
    actions.putAll(2, INSTANCE_UPDATING, INSTANCE_UPDATED);
    assertState(ROLLED_FORWARD, actions.build());

    assertJobState(
        JOB,
        ImmutableMap.of(0, NEW_CONFIG, 1, NEW_CONFIG, 2, NEW_CONFIG));

    try {
      updater.rollback(UPDATE_ID, AUDIT);
      fail();
    } catch (UpdateStateException e) {
      // Expected.
    }
  }

  @Test
  public void testRollbackDuringUpgrade() throws Exception {
    expectTaskKilled().times(5);

    control.replay();

    JobUpdate builder = makeJobUpdate(makeInstanceConfig(0, 2, OLD_CONFIG)).newBuilder();
    builder.getInstructions().getSettings()
        .setUpdateStrategy(
            JobUpdateStrategy.batchStrategy(new BatchJobUpdateStrategy().setGroupSize(2)));
    IJobUpdate update = IJobUpdate.build(builder);
    insertInitialTasks(update);

    changeState(JOB, 0, ASSIGNED, STARTING, RUNNING);
    changeState(JOB, 1, ASSIGNED, STARTING, RUNNING);
    changeState(JOB, 2, ASSIGNED, STARTING, RUNNING);
    clock.advance(WATCH_TIMEOUT);

    ImmutableMultimap.Builder<Integer, JobUpdateAction> actions = ImmutableMultimap.builder();

    // Instances 0 and 1 are updated.
    updater.start(update, AUDIT);
    actions.putAll(0, INSTANCE_UPDATING)
        .putAll(1, INSTANCE_UPDATING);
    assertState(ROLLING_FORWARD, actions.build());
    changeState(JOB, 1, FINISHED, ASSIGNED, STARTING, RUNNING);
    changeState(JOB, 0, FINISHED, ASSIGNED, STARTING, RUNNING);
    clock.advance(WATCH_TIMEOUT);

    actions.putAll(0, INSTANCE_UPDATED)
        .putAll(1, INSTANCE_UPDATED)
        .putAll(2, INSTANCE_UPDATING);
    assertState(ROLLING_FORWARD, actions.build());
    clock.advance(WATCH_TIMEOUT);

    updater.rollback(UPDATE_ID, AUDIT);

    actions.putAll(1, INSTANCE_ROLLING_BACK);
    actions.putAll(2, INSTANCE_ROLLING_BACK);
    changeState(JOB, 1, KILLED);
    changeState(JOB, 2, KILLED);
    clock.advance(WATCH_TIMEOUT);

    assertState(ROLLING_BACK, actions.build());
    clock.advance(WATCH_TIMEOUT);

    changeState(JOB, 2, ASSIGNED, STARTING, RUNNING);
    changeState(JOB, 1, ASSIGNED, STARTING, RUNNING);
    clock.advance(WATCH_TIMEOUT);

    actions.putAll(2, INSTANCE_ROLLED_BACK)
        .putAll(1, INSTANCE_ROLLED_BACK);
    changeState(JOB, 0, KILLED);
    actions.putAll(0, INSTANCE_ROLLING_BACK);
    clock.advance(WATCH_TIMEOUT);

    assertState(ROLLING_BACK, actions.build());

    changeState(JOB, 0, ASSIGNED, STARTING, RUNNING);
    actions.putAll(0, INSTANCE_ROLLED_BACK);
    clock.advance(WATCH_TIMEOUT);

    assertState(ROLLED_BACK, actions.build());

    assertJobState(
        JOB,
        ImmutableMap.of(0, OLD_CONFIG, 1, OLD_CONFIG, 2, OLD_CONFIG));
  }

  @Test
  public void testRollbackCoordinatedUpdate() throws Exception {
    control.replay();

    JobUpdate builder = makeJobUpdate(
        // No-op - task is already matching the new config.
        makeInstanceConfig(0, 0, NEW_CONFIG),
        // Tasks needing update.
        makeInstanceConfig(1, 2, OLD_CONFIG)).newBuilder();

    builder.getInstructions().getSettings().setBlockIfNoPulsesAfterMs((int) PULSE_TIMEOUT_MS);
    insertInitialTasks(IJobUpdate.build(builder));

    changeState(JOB, 0, ASSIGNED, STARTING, RUNNING);
    changeState(JOB, 1, ASSIGNED, STARTING, RUNNING);
    changeState(JOB, 2, ASSIGNED, STARTING, RUNNING);
    clock.advance(WATCH_TIMEOUT);

    ImmutableMultimap.Builder<Integer, JobUpdateAction> actions = ImmutableMultimap.builder();
    updater.start(IJobUpdate.build(builder), AUDIT);

    // The update is blocked initially waiting for a pulse.
    assertState(ROLL_FORWARD_AWAITING_PULSE, actions.build());

    updater.rollback(UPDATE_ID, AUDIT);

    clock.advance(WATCH_TIMEOUT);
    assertState(ROLLED_BACK, actions.build());
  }

  @Test
  public void testRollbackPausedForwardUpdate() throws Exception {
    expectTaskKilled().times(2);

    control.replay();

    JobUpdate builder = makeJobUpdate(
        // No-op - task is already matching the new config.
        makeInstanceConfig(0, 0, NEW_CONFIG),
        // Tasks needing update.
        makeInstanceConfig(1, 2, OLD_CONFIG)).newBuilder();

    insertInitialTasks(IJobUpdate.build(builder));

    changeState(JOB, 0, ASSIGNED, STARTING, RUNNING);
    changeState(JOB, 1, ASSIGNED, STARTING, RUNNING);
    changeState(JOB, 2, ASSIGNED, STARTING, RUNNING);
    clock.advance(WATCH_TIMEOUT);

    ImmutableMultimap.Builder<Integer, JobUpdateAction> actions = ImmutableMultimap.builder();
    updater.start(IJobUpdate.build(builder), AUDIT);

    actions.putAll(1, INSTANCE_UPDATING);
    assertState(ROLLING_FORWARD, actions.build());
    clock.advance(WATCH_TIMEOUT);
    changeState(JOB, 1, KILLED, ASSIGNED, STARTING, RUNNING);

    updater.pause(UPDATE_ID, AUDIT);
    assertState(ROLL_FORWARD_PAUSED, actions.build());

    updater.rollback(UPDATE_ID, AUDIT);

    actions.putAll(1, INSTANCE_ROLLING_BACK);
    clock.advance(WATCH_TIMEOUT);
    assertState(ROLLING_BACK, actions.build());

    actions.putAll(1, INSTANCE_ROLLED_BACK);
    changeState(JOB, 1, KILLED, ASSIGNED, STARTING, RUNNING);
    clock.advance(WATCH_TIMEOUT);
    assertState(ROLLED_BACK, actions.build());

    assertJobState(
        JOB,
        ImmutableMap.of(0, NEW_CONFIG, 1, OLD_CONFIG, 2, OLD_CONFIG));
  }

  @Test
  public void testInProgressUpdate() throws Exception {
    control.replay();

    IJobUpdate inProgress = makeJobUpdate();
    storage.write((NoResult.Quiet) storeProvider ->
        saveJobUpdate(storeProvider.getJobUpdateStore(), inProgress, ROLLING_FORWARD));
    IJobUpdate anotherUpdate = makeJobUpdate();
    try {
      updater.start(anotherUpdate, AUDIT);
      fail("update cannot start when another is in-progress");
    } catch (UpdateInProgressException e) {
      // Expected.
      assertEquals(
          inProgress.getSummary().newBuilder().setState(new JobUpdateState(ROLLING_FORWARD, 0, 0)),
          e.getInProgressUpdateSummary().newBuilder());
    }
  }

  @Test
  public void testSuccessfulSlaAwareUpdate() throws Exception {
    expectTaskKilled().times(3);

    control.replay();

    // Our batch size is 3 but our SLA policy only allows for 1 instance to be down at a time.
    // We want to ensure that only one instance is killed at a time.
    JobUpdate builder = makeJobUpdate(3, SLA_AWARE_CONFIG, makeInstanceConfig(0, 2, OLD_CONFIG))
        .newBuilder();
    builder.getInstructions().getSettings().setSlaAware(true);
    IJobUpdate update = setInstanceCount(IJobUpdate.build(builder), 3);
    insertInitialTasks(update);

    changeState(JOB, 0, ASSIGNED, STARTING, RUNNING);
    changeState(JOB, 1, ASSIGNED, STARTING, RUNNING);
    changeState(JOB, 2, ASSIGNED, STARTING, RUNNING);

    ImmutableMultimap.Builder<Integer, JobUpdateAction> actions = ImmutableMultimap.builder();

    updater.start(update, AUDIT);
    actions.put(0, INSTANCE_UPDATING);
    actions.put(0, INSTANCE_UPDATING); // Awaiting SLA check to pass
    actions.put(1, INSTANCE_UPDATING);
    actions.put(1, INSTANCE_UPDATING); // Awaiting SLA check to pass
    actions.put(2, INSTANCE_UPDATING);
    actions.put(2, INSTANCE_UPDATING); // Awaiting SLA check to pass

    // SLA aware update should only send one instance to KILLING -- find which one.
    int firstTaskBeingKilled = Iterables
        .getOnlyElement(getTasksInState(JOB, KILLING))
        .getAssignedTask()
        .getInstanceId();
    actions.put(firstTaskBeingKilled, INSTANCE_UPDATING); // SLA check passed, killing
    changeState(JOB, firstTaskBeingKilled, KILLED);

    // Sanity check that after a kill delay but before the task is RUNNING again, we do not kill
    // another task.
    clock.advance(WATCH_TIMEOUT);
    assertTrue(Iterables.isEmpty(getTasksInState(JOB, KILLING)));
    assertState(ROLLING_FORWARD, actions.build());

    // First task finishes updating
    changeState(JOB, firstTaskBeingKilled, ASSIGNED, STARTING, RUNNING);
    clock.advance(WATCH_TIMEOUT);
    actions.put(firstTaskBeingKilled, INSTANCE_UPDATED);

    // Update the second task
    int secondTaskBeingKilled = Iterables
        .getOnlyElement(getTasksInState(JOB, KILLING))
        .getAssignedTask()
        .getInstanceId();
    actions.put(secondTaskBeingKilled, INSTANCE_UPDATING); // SLA check passed, killing
    changeState(JOB, secondTaskBeingKilled, KILLED);
    changeState(JOB, secondTaskBeingKilled, ASSIGNED, STARTING, RUNNING);
    clock.advance(WATCH_TIMEOUT);
    actions.put(secondTaskBeingKilled, INSTANCE_UPDATED);

    // Update the final task
    int finalTaskBeingKilled = Iterables
        .getOnlyElement(getTasksInState(JOB, KILLING))
        .getAssignedTask()
        .getInstanceId();
    actions.put(finalTaskBeingKilled, INSTANCE_UPDATING); // SLA check passed, killing
    changeState(JOB, finalTaskBeingKilled, KILLED);
    changeState(JOB, finalTaskBeingKilled, ASSIGNED, STARTING, RUNNING);
    clock.advance(WATCH_TIMEOUT);
    actions.put(finalTaskBeingKilled, INSTANCE_UPDATED);

    assertState(ROLLED_FORWARD, actions.build());
    assertJobState(
        JOB,
        ImmutableMap.of(0, SLA_AWARE_CONFIG, 1, SLA_AWARE_CONFIG, 2, SLA_AWARE_CONFIG));
  }

  @Test
  public void testSuccessfulSlaAwareUpdateWithPause() throws Exception {
    expectTaskKilled().times(3);

    control.replay();

    JobUpdate builder = makeJobUpdate(3, SLA_AWARE_CONFIG, makeInstanceConfig(0, 2, OLD_CONFIG))
        .newBuilder();
    builder.getInstructions().getSettings().setSlaAware(true);
    IJobUpdate update = setInstanceCount(IJobUpdate.build(builder), 3);
    insertInitialTasks(update);

    changeState(JOB, 0, ASSIGNED, STARTING, RUNNING);
    changeState(JOB, 1, ASSIGNED, STARTING, RUNNING);
    changeState(JOB, 2, ASSIGNED, STARTING, RUNNING);

    ImmutableMultimap.Builder<Integer, JobUpdateAction> actions = ImmutableMultimap.builder();

    updater.start(update, AUDIT);
    actions.put(0, INSTANCE_UPDATING);
    actions.put(0, INSTANCE_UPDATING);
    actions.put(1, INSTANCE_UPDATING);
    actions.put(1, INSTANCE_UPDATING);
    actions.put(2, INSTANCE_UPDATING);
    actions.put(2, INSTANCE_UPDATING);

    int firstTaskBeingKilled = Iterables
        .getOnlyElement(getTasksInState(JOB, KILLING))
        .getAssignedTask()
        .getInstanceId();
    actions.put(firstTaskBeingKilled, INSTANCE_UPDATING);
    changeState(JOB, firstTaskBeingKilled, KILLED);
    changeState(JOB, firstTaskBeingKilled, ASSIGNED, STARTING, RUNNING);

    // Pause the update to stop progress
    assertState(ROLLING_FORWARD, actions.build());
    updater.pause(UPDATE_ID, AUDIT);
    assertState(ROLL_FORWARD_PAUSED, actions.build());

    // Ensure no tasks are killed while paused
    clock.advance(WATCH_TIMEOUT);
    assertTrue(Iterables.isEmpty(getTasksInState(JOB, KILLING)));

    // Unpause and continue as normal
    updater.resume(UPDATE_ID, AUDIT);
    clock.advance(WATCH_TIMEOUT);
    actions.put(firstTaskBeingKilled, INSTANCE_UPDATED);

    int secondTaskBeingKilled = Iterables
        .getOnlyElement(getTasksInState(JOB, KILLING))
        .getAssignedTask()
        .getInstanceId();
    actions.put(secondTaskBeingKilled, INSTANCE_UPDATING);
    changeState(JOB, secondTaskBeingKilled, KILLED);
    changeState(JOB, secondTaskBeingKilled, ASSIGNED, STARTING, RUNNING);
    clock.advance(WATCH_TIMEOUT);
    actions.put(secondTaskBeingKilled, INSTANCE_UPDATED);

    int finalTaskBeingKilled = Iterables
        .getOnlyElement(getTasksInState(JOB, KILLING))
        .getAssignedTask()
        .getInstanceId();
    actions.put(finalTaskBeingKilled, INSTANCE_UPDATING);
    changeState(JOB, finalTaskBeingKilled, KILLED);
    changeState(JOB, finalTaskBeingKilled, ASSIGNED, STARTING, RUNNING);
    clock.advance(WATCH_TIMEOUT);
    actions.put(finalTaskBeingKilled, INSTANCE_UPDATED);

    assertState(ROLLED_FORWARD, actions.build());
    assertJobState(
        JOB,
        ImmutableMap.of(0, SLA_AWARE_CONFIG, 1, SLA_AWARE_CONFIG, 2, SLA_AWARE_CONFIG));
  }

  @Test
  public void testSuccessfulSlaAwareUpdateWithRollback() throws Exception {
    expectTaskKilled().times(4);
    // We need both the old and new config to be SLA aware for this rollback
    ITaskConfig slaAwareOldConfig = setCountSlaPolicy(setExecutorData(OLD_CONFIG, "old"), 2, 0);

    control.replay();

    JobUpdate builder = makeJobUpdate(
        3,
        SLA_AWARE_CONFIG,
        makeInstanceConfig(0, 2, slaAwareOldConfig)).newBuilder();
    builder.getInstructions().getSettings().setSlaAware(true);
    IJobUpdate update = setInstanceCount(IJobUpdate.build(builder), 3);
    insertInitialTasks(update);

    changeState(JOB, 0, ASSIGNED, STARTING, RUNNING);
    changeState(JOB, 1, ASSIGNED, STARTING, RUNNING);
    changeState(JOB, 2, ASSIGNED, STARTING, RUNNING);

    ImmutableMultimap.Builder<Integer, JobUpdateAction> actions = ImmutableMultimap.builder();

    updater.start(update, AUDIT);
    actions.put(0, INSTANCE_UPDATING);
    actions.put(0, INSTANCE_UPDATING);
    actions.put(1, INSTANCE_UPDATING);
    actions.put(1, INSTANCE_UPDATING);
    actions.put(2, INSTANCE_UPDATING);
    actions.put(2, INSTANCE_UPDATING);

    int firstTaskBeingKilled = Iterables
        .getOnlyElement(getTasksInState(JOB, KILLING))
        .getAssignedTask()
        .getInstanceId();
    actions.put(firstTaskBeingKilled, INSTANCE_UPDATING);
    changeState(JOB, firstTaskBeingKilled, KILLED);
    changeState(JOB, firstTaskBeingKilled, ASSIGNED, STARTING, RUNNING);
    clock.advance(WATCH_TIMEOUT);
    actions.put(firstTaskBeingKilled, INSTANCE_UPDATED);

    int secondTaskBeingKilled = Iterables
        .getOnlyElement(getTasksInState(JOB, KILLING))
        .getAssignedTask()
        .getInstanceId();
    actions.put(secondTaskBeingKilled, INSTANCE_UPDATING);
    changeState(JOB, secondTaskBeingKilled, KILLED);
    changeState(JOB, secondTaskBeingKilled, ASSIGNED, STARTING, RUNNING);

    int finalTaskNotKilled = Iterables.getOnlyElement(getTasksInState(JOB, RUNNING)
        .stream()
        .filter(t -> t.getAssignedTask().getTask().equals(slaAwareOldConfig))
        .map(t -> t.getAssignedTask().getInstanceId())
        .collect(Collectors.toList()));

    // Roll the update back with 2/3 instances on the new config
    assertJobState(
        JOB,
        ImmutableMap.of(
            firstTaskBeingKilled, SLA_AWARE_CONFIG,
            secondTaskBeingKilled, SLA_AWARE_CONFIG,
            finalTaskNotKilled, slaAwareOldConfig));
    updater.rollback(UPDATE_ID, AUDIT);
    actions.put(firstTaskBeingKilled, INSTANCE_ROLLING_BACK);
    actions.put(firstTaskBeingKilled, INSTANCE_ROLLING_BACK); // Awaiting SLA check to pass
    actions.put(secondTaskBeingKilled, INSTANCE_ROLLING_BACK);
    actions.put(secondTaskBeingKilled, INSTANCE_ROLLING_BACK); // Awaiting SLA check to pass
    actions.put(finalTaskNotKilled, INSTANCE_ROLLING_BACK);
    actions.put(finalTaskNotKilled, INSTANCE_ROLLED_BACK); // Task never updated originally

    int firstRollbackTaskBeingKilled = Iterables
        .getOnlyElement(getTasksInState(JOB, KILLING))
        .getAssignedTask()
        .getInstanceId();
    actions.put(firstRollbackTaskBeingKilled, INSTANCE_ROLLING_BACK); // SLA check passed
    changeState(JOB, firstRollbackTaskBeingKilled, KILLED);
    changeState(JOB, firstRollbackTaskBeingKilled, ASSIGNED, STARTING, RUNNING);
    clock.advance(WATCH_TIMEOUT);
    actions.put(firstRollbackTaskBeingKilled, INSTANCE_ROLLED_BACK);

    int secondRollbackTaskBeingKilled = Iterables
        .getOnlyElement(getTasksInState(JOB, KILLING))
        .getAssignedTask()
        .getInstanceId();
    actions.put(secondRollbackTaskBeingKilled, INSTANCE_ROLLING_BACK); // SLA check passed
    changeState(JOB, secondRollbackTaskBeingKilled, KILLED);
    changeState(JOB, secondRollbackTaskBeingKilled, ASSIGNED, STARTING, RUNNING);
    clock.advance(WATCH_TIMEOUT);
    actions.put(secondRollbackTaskBeingKilled, INSTANCE_ROLLED_BACK);

    assertState(ROLLED_BACK, actions.build());
    assertJobState(
        JOB,
        ImmutableMap.of(0, slaAwareOldConfig, 1, slaAwareOldConfig, 2, slaAwareOldConfig));
  }

  @Test
  public void testSuccessfulBatchUpdateAutoPause() throws Exception {
    expectTaskKilled().times(3);

    control.replay();

    JobUpdate builder = makeJobUpdate(makeInstanceConfig(0, 2, OLD_CONFIG)).newBuilder();
    builder.getInstructions().getSettings()
        .setUpdateStrategy(
            JobUpdateStrategy.batchStrategy(
                new BatchJobUpdateStrategy().setGroupSize(2).setAutopauseAfterBatch(true)));
    IJobUpdate update = IJobUpdate.build(builder);
    insertInitialTasks(update);

    changeState(JOB, 0, ASSIGNED, STARTING, RUNNING);
    changeState(JOB, 1, ASSIGNED, STARTING, RUNNING);
    changeState(JOB, 2, ASSIGNED, STARTING, RUNNING);
    clock.advance(WATCH_TIMEOUT);

    ImmutableMultimap.Builder<Integer, JobUpdateAction> actions = ImmutableMultimap.builder();

    // Instances 0 and 1 are updated.
    updater.start(update, AUDIT);
    actions.put(0, INSTANCE_UPDATING).putAll(1, INSTANCE_UPDATING);
    assertState(ROLLING_FORWARD, actions.build());
    changeState(JOB, 0, FINISHED, ASSIGNED, STARTING, RUNNING);
    changeState(JOB, 1, FINISHED, ASSIGNED, STARTING, RUNNING);
    clock.advance(WATCH_TIMEOUT);

    actions.put(0, INSTANCE_UPDATED).put(1, INSTANCE_UPDATED);

    // Update should now be paused
    assertState(ROLL_FORWARD_PAUSED, actions.build());

    // Continue the update
    updater.resume(UPDATE_ID, AUDIT);

    actions.put(2, INSTANCE_UPDATING);

    assertState(ROLLING_FORWARD, actions.build());

    // Instance 2 is updated.
    changeState(JOB, 2, FINISHED, ASSIGNED, STARTING, RUNNING);
    clock.advance(WATCH_TIMEOUT);
    actions.put(2, INSTANCE_UPDATED);

    assertState(ROLLED_FORWARD, actions.build());

    assertJobState(
        JOB,
        ImmutableMap.of(0, NEW_CONFIG, 1, NEW_CONFIG, 2, NEW_CONFIG));
  }

  @Test
  public void testSuccessfulVarBatchUpdateAutoPause() throws Exception {
    expectTaskKilled().times(6);

    control.replay();

    JobUpdate builder = makeJobUpdate(makeInstanceConfig(0, 5, OLD_CONFIG)).newBuilder();
    builder.getInstructions().getSettings()
        .setUpdateStrategy(
            JobUpdateStrategy.varBatchStrategy(
                new VariableBatchJobUpdateStrategy()
                    .setGroupSizes(ImmutableList.of(1, 2, 3))
                    .setAutopauseAfterBatch(true)));
    IJobUpdate update = setInstanceCount(IJobUpdate.build(builder), 6);
    insertInitialTasks(update);

    for (int i = 0; i <= 5; ++i) {
      changeState(JOB, i, ASSIGNED, STARTING, RUNNING);
    }
    clock.advance(WATCH_TIMEOUT);

    ImmutableMultimap.Builder<Integer, JobUpdateAction> actions = ImmutableMultimap.builder();

    // First batch is updated
    updater.start(update, AUDIT);
    actions.put(0, INSTANCE_UPDATING);
    assertState(ROLLING_FORWARD, actions.build());
    changeState(JOB, 0, FINISHED, ASSIGNED, STARTING, RUNNING);
    clock.advance(WATCH_TIMEOUT);

    actions.put(0, INSTANCE_UPDATED);

    // Update should now be paused after first batch is done.
    assertState(ROLL_FORWARD_PAUSED, actions.build());
    updater.resume(UPDATE_ID, AUDIT);

    actions.put(1, INSTANCE_UPDATING).put(2, INSTANCE_UPDATING);
    assertState(ROLLING_FORWARD, actions.build());

    // Second batch is moving forward.
    // Make instance 1 the instance that waits for final transition to SUCCEED
    changeState(JOB, 2, FINISHED, ASSIGNED, STARTING, RUNNING);
    clock.advance(Amount.of(WATCH_TIMEOUT.getValue() / 2, Time.MILLISECONDS));
    changeState(JOB, 1, FINISHED, ASSIGNED, STARTING, RUNNING);
    clock.advance(WATCH_TIMEOUT);

    actions.put(1, INSTANCE_UPDATED);
    actions.put(2, INSTANCE_UPDATED);

    // Second autoPause at second barrier
    assertState(ROLL_FORWARD_PAUSED, actions.build());

    updater.resume(UPDATE_ID, AUDIT);
    actions.put(3, INSTANCE_UPDATING).put(4, INSTANCE_UPDATING).put(5, INSTANCE_UPDATING);

    assertState(ROLLING_FORWARD, actions.build());

    // Third batch is moving forward.
    changeState(JOB, 3, FINISHED, ASSIGNED, STARTING, RUNNING);
    changeState(JOB, 4, FINISHED, ASSIGNED, STARTING, RUNNING);
    changeState(JOB, 5, FINISHED, ASSIGNED, STARTING, RUNNING);
    clock.advance(WATCH_TIMEOUT);

    actions.put(3, INSTANCE_UPDATED).put(4, INSTANCE_UPDATED).put(5, INSTANCE_UPDATED);

    assertState(ROLLED_FORWARD, actions.build());

    assertJobState(
        JOB,
        ImmutableMap.<Integer, ITaskConfig>builder()
            .put(0, NEW_CONFIG)
            .put(1, NEW_CONFIG)
            .put(2, NEW_CONFIG)
            .put(3, NEW_CONFIG)
            .put(4, NEW_CONFIG)
            .put(5, NEW_CONFIG)
            .build());
  }

  @Test
  public void testSuccessfulVarBatchUpdateAutoPauseWithRollback() throws Exception {
    expectTaskKilled().times(4);

    control.replay();

    // This update will add an instances and update two instances
    JobUpdate builder = makeJobUpdate(makeInstanceConfig(0, 1, OLD_CONFIG)).newBuilder();
    builder.getInstructions().getSettings()
        .setMaxFailedInstances(0)
        .setUpdateStrategy(
            JobUpdateStrategy.varBatchStrategy(
                new VariableBatchJobUpdateStrategy()
                    .setGroupSizes(ImmutableList.of(1, 2))
                    .setAutopauseAfterBatch(true)));
    IJobUpdate update = setInstanceCount(IJobUpdate.build(builder), 3);
    insertInitialTasks(update);

    // Only have two instances so that one has to be added to test the add instance feature
    // of this strategy
    changeState(JOB, 0, ASSIGNED, STARTING, RUNNING);
    changeState(JOB, 1, ASSIGNED, STARTING, RUNNING);

    clock.advance(WATCH_TIMEOUT);

    ImmutableMultimap.Builder<Integer, JobUpdateAction> actions = ImmutableMultimap.builder();

    // First batch is updated, Instance will be created
    updater.start(update, AUDIT);
    actions.put(2, INSTANCE_UPDATING);
    assertState(ROLLING_FORWARD, actions.build());
    changeState(JOB, 2, ASSIGNED, STARTING, RUNNING);
    clock.advance(WATCH_TIMEOUT);

    actions.put(2, INSTANCE_UPDATED);

    // Update should now be paused after first batch is done.
    assertState(ROLL_FORWARD_PAUSED, actions.build());
    updater.resume(UPDATE_ID, AUDIT);

    actions.put(0, INSTANCE_UPDATING).put(1, INSTANCE_UPDATING);
    assertState(ROLLING_FORWARD, actions.build());

    // Move on to batch two
    changeState(JOB, 0, KILLED, ASSIGNED, STARTING, RUNNING);
    clock.advance(WATCH_TIMEOUT);

    // Trigger a rollback job half way through by failing task 1
    changeState(JOB, 1, FINISHED, ASSIGNED, STARTING, RUNNING);
    clock.advance(FLAPPING_THRESHOLD);
    changeState(JOB, 1, FAILED);

    actions.putAll(0, INSTANCE_UPDATED, INSTANCE_ROLLING_BACK)
        .putAll(1, INSTANCE_UPDATE_FAILED, INSTANCE_ROLLING_BACK);

    assertState(ROLLING_BACK, actions.build());
    assertLatestUpdateMessage(JobUpdateControllerImpl.failureMessage(1, Failure.EXITED));

    // Rollback Instances 0 and 1
    changeState(JOB, 0, KILLED, ASSIGNED, STARTING, RUNNING);
    changeState(JOB, 1, ASSIGNED, STARTING, RUNNING);
    clock.advance(WATCH_TIMEOUT);

    actions.put(0, INSTANCE_ROLLED_BACK).put(1, INSTANCE_ROLLED_BACK).put(2, INSTANCE_ROLLING_BACK);
    assertState(ROLLING_BACK, actions.build());

    // Rollback instance 2
    changeState(JOB, 2, KILLED);
    actions.put(2, INSTANCE_ROLLED_BACK);

    clock.advance(WATCH_TIMEOUT);

    assertState(ROLLED_BACK, actions.build());

    assertJobState(
        JOB,
        ImmutableMap.<Integer, ITaskConfig>builder()
            .put(0, OLD_CONFIG)
            .put(1, OLD_CONFIG)
            .build());
  }

  private Collection<IScheduledTask> getTasksInState(IJobKey job, ScheduleStatus status) {
    return storage.write(storeProvider ->
        storeProvider.getTaskStore().fetchTasks(Query.jobScoped(job).byStatus(status)));
  }

  private static IJobUpdateSummary makeUpdateSummary(IJobUpdateKey key) {
    return IJobUpdateSummary.build(new JobUpdateSummary()
        .setUser("user")
        .setKey(key.newBuilder()));
  }

  private static IJobUpdate makeJobUpdate(IInstanceTaskConfig... configs) {
    return makeJobUpdate(1, NEW_CONFIG, configs);
  }

  private static IJobUpdate makeJobUpdate(int updateGroupSize,
                                          ITaskConfig newConfig,
                                          IInstanceTaskConfig... configs) {

    JobUpdate builder = new JobUpdate()
        .setSummary(makeUpdateSummary(UPDATE_ID).newBuilder().setMetadata(METADATA))
        .setInstructions(new JobUpdateInstructions()
            .setDesiredState(new InstanceTaskConfig()
                .setTask(newConfig.newBuilder())
                .setInstances(ImmutableSet.of(new Range(0, 2))))
            .setSettings(new JobUpdateSettings()
                .setUpdateStrategy(
                    JobUpdateStrategy.queueStrategy(
                        new QueueJobUpdateStrategy().setGroupSize(updateGroupSize)))
                .setRollbackOnFailure(true)
                .setMinWaitInInstanceRunningMs(WATCH_TIMEOUT.as(Time.MILLISECONDS).intValue())
                .setUpdateOnlyTheseInstances(ImmutableSet.of())));

    for (IInstanceTaskConfig config : configs) {
      builder.getInstructions().addToInitialState(config.newBuilder());
    }

    return IJobUpdate.build(builder);
  }

  private static IJobUpdate setInstanceCount(IJobUpdate update, int instanceCount) {
    JobUpdate builder = update.newBuilder();
    builder.getInstructions().getDesiredState().setInstances(
        ImmutableSet.of(new Range(0, instanceCount - 1)));
    return IJobUpdate.build(builder);
  }

  private static IInstanceTaskConfig makeInstanceConfig(int start, int end, ITaskConfig config) {
    return IInstanceTaskConfig.build(new InstanceTaskConfig()
        .setInstances(ImmutableSet.of(new Range(start, end)))
        .setTask(config.newBuilder()));
  }
}