StateManagerImpl implements StateManager { @Override public void insertPendingTasks( MutableStoreProvider storeProvider, final ITaskConfig task, Set<Integer> instanceIds) { requireNonNull(storeProvider); requireNonNull(task); checkNotBlank(instanceIds); Set<IScheduledTask> scheduledTasks = FluentIterable.from(instanceIds) .transform(instanceId -> createTask(instanceId, task)).toSet(); Iterable<IScheduledTask> existingTasks = storeProvider.getTaskStore().fetchTasks( Query.jobScoped(task.getJob()).active()); Set<Integer> existingInstanceIds = FluentIterable.from(existingTasks).transform(Tasks::getInstanceId).toSet(); if (!Sets.intersection(existingInstanceIds, instanceIds).isEmpty()) { throw new IllegalArgumentException("Instance ID collision detected."); } storeProvider.getUnsafeTaskStore().saveTasks(scheduledTasks); for (IScheduledTask scheduledTask : scheduledTasks) { updateTaskAndExternalState( storeProvider.getUnsafeTaskStore(), Tasks.id(scheduledTask), Optional.of(scheduledTask), Optional.of(PENDING), Optional.absent()); } } @Inject  StateManagerImpl(
      final Clock clock,
      Driver driver,
      TaskIdGenerator taskIdGenerator,
      EventSink eventSink,
      RescheduleCalculator rescheduleCalculator); @Override void insertPendingTasks(
      MutableStoreProvider storeProvider,
      final ITaskConfig task,
      Set<Integer> instanceIds); @Override StateChangeResult changeState(
      MutableStoreProvider storeProvider,
      String taskId,
      Optional<ScheduleStatus> casState,
      final ScheduleStatus newState,
      final Optional<String> auditMessage); @Override IAssignedTask assignTask(
      MutableStoreProvider storeProvider,
      String taskId,
      String slaveHost,
      SlaveID slaveId,
      Map<String, Integer> assignedPorts); @Override void deleteTasks(MutableStoreProvider storeProvider, final Set<String> taskIds);  }
@Test(expected = IllegalArgumentException.class) public void insertEmptyPendingInstancesFails() { control.replay(); storage.write((NoResult.Quiet) storeProvider -> stateManager.insertPendingTasks( storeProvider, NON_SERVICE_CONFIG, ImmutableSet.of())); }