StorageRecovery { public void run() { setup(); LOG.info("start recovering..."); systemAdmins.start(); this.containerStorageManagers.forEach((containerName, containerStorageManager) -> { try { containerStorageManager.start(); } catch (InterruptedException e) { LOG.warn("Received an interrupt during store restoration for container {}." + " Proceeding with the next container", containerName); } }); this.containerStorageManagers.forEach((containerName, containerStorageManager) -> containerStorageManager.shutdown()); systemAdmins.stop(); LOG.info("successfully recovered in " + storeBaseDir.toString()); } StorageRecovery(Config config, String path); void run();  }
@Test public void testStorageEngineReceivedAllValues() { MockCoordinatorStreamSystemFactory.enableMockConsumerCache(); StorageRecovery storageRecovery = new StorageRecovery(config, path); storageRecovery.run(); assertEquals(2, MockStorageEngine.incomingMessageEnvelopes.size()); assertEquals(msg, MockStorageEngine.incomingMessageEnvelopes.get(0)); assertEquals(msg, MockStorageEngine.incomingMessageEnvelopes.get(1)); String expectedStoreDir = String.format("%s/state/%s/Partition_", path, STORE_NAME); String actualStoreDir = MockStorageEngine.storeDir.toString(); assertEquals(expectedStoreDir, actualStoreDir.substring(0, actualStoreDir.length() - 1)); }