LogStorage implements NonVolatileStorage, DistributedSnapshotStore { @Override public synchronized void start(final MutateWork.NoResult.Quiet initializationLogic) { write((NoResult.Quiet) unused -> { recover(); migrate(); recovered = true; write(initializationLogic); }); scheduleSnapshots(); } @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 testStart() throws Exception { expect(log.open()).andReturn(stream); AtomicBoolean initialized = new AtomicBoolean(false); MutateWork.NoResult.Quiet initializationLogic = provider -> { initialized.set(true); }; Capture<MutateWork.NoResult.Quiet> recoverAndInitializeWork = createCapture(); storageUtil.storage.write(capture(recoverAndInitializeWork)); expectLastCall().andAnswer(() -> { recoverAndInitializeWork.getValue().apply(storageUtil.mutableStoreProvider); return null; }); Capture<MutateWork<Void, RuntimeException>> recoveryWork = createCapture(); expect(storageUtil.storage.write(capture(recoveryWork))).andAnswer( () -> { recoveryWork.getValue().apply(storageUtil.mutableStoreProvider); return null; }); Capture<MutateWork<Void, RuntimeException>> initializationWork = createCapture(); expect(storageUtil.storage.write(capture(initializationWork))).andAnswer( () -> { initializationWork.getValue().apply(storageUtil.mutableStoreProvider); return null; }); Capture<Runnable> snapshotAction = createCapture(); schedulingService.doEvery(eq(SNAPSHOT_INTERVAL), capture(snapshotAction)); Snapshot snapshotContents = new Snapshot() .setTimestamp(NOW) .setTasks(ImmutableSet.of( TaskTestUtil.makeTask("task_id", TaskTestUtil.JOB).newBuilder())); expect(snapshotStore.createSnapshot()).andReturn(snapshotContents); DeduplicatedSnapshot deduplicated = new SnapshotDeduplicatorImpl().deduplicate(snapshotContents); expect(deduplicator.deduplicate(snapshotContents)).andReturn(deduplicated); streamMatcher.expectSnapshot(deduplicated).andReturn(position); stream.truncateBefore(position); Capture<MutateWork<Void, RuntimeException>> snapshotWork = createCapture(); expect(storageUtil.storage.write(capture(snapshotWork))).andAnswer( () -> { snapshotWork.getValue().apply(storageUtil.mutableStoreProvider); return null; }).anyTimes(); buildReplayLogEntries(); control.replay(); logStorage.prepare(); logStorage.start(initializationLogic); assertTrue(initialized.get()); snapshotAction.getValue().run(); assertEquals( Sets.complementOf(EnumSet.of( LogEntry._Fields.FRAME, LogEntry._Fields.DEDUPLICATED_SNAPSHOT, LogEntry._Fields.DEFLATED_ENTRY)), EnumSet.copyOf(logStorage.buildLogEntryReplayActions().keySet())); assertEquals( EnumSet.allOf(Op._Fields.class), EnumSet.copyOf(logStorage.buildTransactionReplayActions().keySet())); }