LogStorage implements NonVolatileStorage, DistributedSnapshotStore { @Override public <T, E extends Exception> T write(final MutateWork<T, E> work) throws StorageException, E { long waitStart = System.nanoTime(); writeLock.lock(); try { writerWaitStats.accumulate(System.nanoTime() - waitStart); if (!recovered) { return writeBehindStorage.write(work); } return doInTransaction(work); } finally { writeLock.unlock(); } } @Inject  LogStorage(
      LogManager logManager,
      ShutdownRegistry shutdownRegistry,
      Settings settings,
      SnapshotStore<Snapshot> snapshotStore,
      @Volatile Storage storage,
      @Volatile SchedulerStore.Mutable schedulerStore,
      @Volatile CronJobStore.Mutable jobStore,
      @Volatile TaskStore.Mutable taskStore,
      @Volatile LockStore.Mutable lockStore,
      @Volatile QuotaStore.Mutable quotaStore,
      @Volatile AttributeStore.Mutable attributeStore,
      @Volatile JobUpdateStore.Mutable jobUpdateStore,
      EventSink eventSink,
      ReentrantLock writeLock,
      MigrationManager migrationManager); @VisibleForTesting  LogStorage(
      LogManager logManager,
      SchedulingService schedulingService,
      SnapshotStore<Snapshot> snapshotStore,
      Amount<Long, Time> snapshotInterval,
      Storage delegateStorage,
      SchedulerStore.Mutable schedulerStore,
      CronJobStore.Mutable jobStore,
      TaskStore.Mutable taskStore,
      LockStore.Mutable lockStore,
      QuotaStore.Mutable quotaStore,
      AttributeStore.Mutable attributeStore,
      JobUpdateStore.Mutable jobUpdateStore,
      EventSink eventSink,
      ReentrantLock writeLock,
      MigrationManager migrationManager); @Override synchronized void prepare(); @Override synchronized void start(final MutateWork.NoResult.Quiet initializationLogic); @Override void stop(); @Timed("scheduler_log_snapshot_persist") @Override void persist(Snapshot snapshot); @Override T write(final MutateWork<T, E> work); @Override T read(Work<T, E> work); @Override void snapshot();  }
@Test public void testNestedTransactions() throws Exception { String taskId = "fred"; Function<IScheduledTask, IScheduledTask> mutation = Functions.identity(); Optional<IScheduledTask> mutated = Optional.of(task("a", ScheduleStatus.STARTING)); ImmutableSet<String> tasksToRemove = ImmutableSet.of("b"); new AbstractMutationFixture() { @Override protected void setupExpectations() throws Exception { storageUtil.expectWrite(); expect(storageUtil.taskStore.mutateTask(taskId, mutation)).andReturn(mutated); storageUtil.taskStore.deleteTasks(tasksToRemove); streamMatcher.expectTransaction( Op.saveTasks(new SaveTasks(ImmutableSet.of(mutated.get().newBuilder()))), Op.removeTasks(new RemoveTasks(tasksToRemove))) .andReturn(position); } @Override protected void performMutations(MutableStoreProvider storeProvider) { assertEquals(mutated, storeProvider.getUnsafeTaskStore().mutateTask(taskId, mutation)); logStorage.write((NoResult.Quiet) innerProvider -> innerProvider.getUnsafeTaskStore().deleteTasks(tasksToRemove)); } }.run(); }