TaskSideInputStorageManager { public void init() { LOG.info("Initializing side input stores."); initializeStoreDirectories(); } TaskSideInputStorageManager(TaskName taskName, TaskMode taskMode, File storeBaseDir, Map<String, StorageEngine> sideInputStores,
      Map<String, Set<SystemStreamPartition>> storesToSSPs, Clock clock); TaskName getTaskName(); void init(); void flush(Map<SystemStreamPartition, String> lastProcessedOffsets); void stop(Map<SystemStreamPartition, String> lastProcessedOffsets); StorageEngine getStore(String storeName); void writeFileOffsets(Map<SystemStreamPartition, String> lastProcessedOffsets); Map<SystemStreamPartition, String> getFileOffsets();  }
@Test public void testInit() { final String storeName = "test-init-store"; final String taskName = "test-init-task"; TaskSideInputStorageManager testSideInputStorageManager = new MockTaskSideInputStorageManagerBuilder(taskName, LOGGED_STORE_DIR) .addLoggedStore(storeName, ImmutableSet.of()) .build(); initializeSideInputStorageManager(testSideInputStorageManager); File storeDir = testSideInputStorageManager.getStoreLocation(storeName); assertTrue("Store directory: " + storeDir.getPath() + " is missing.", storeDir.exists()); }