TaskSideInputStorageManager { public void flush(Map<SystemStreamPartition, String> lastProcessedOffsets) { LOG.info("Flushing the side input stores."); stores.values().forEach(StorageEngine::flush); writeFileOffsets(lastProcessedOffsets); } 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 testFlush() { final String storeName = "test-flush-store"; final String taskName = "test-flush-task"; final SystemStreamPartition ssp = new SystemStreamPartition("test-system", "test-stream", new Partition(0)); final String offset = "123"; final ImmutableMap<SystemStreamPartition, String> processedOffsets = ImmutableMap.of(ssp, offset); TaskSideInputStorageManager testSideInputStorageManager = new MockTaskSideInputStorageManagerBuilder(taskName, LOGGED_STORE_DIR) .addLoggedStore(storeName, ImmutableSet.of(ssp)) .build(); Map<String, StorageEngine> stores = new HashMap<>(); initializeSideInputStorageManager(testSideInputStorageManager); testSideInputStorageManager.flush(processedOffsets); for (StorageEngine storageEngine : stores.values()) { verify(storageEngine).flush(); } verify(testSideInputStorageManager).writeFileOffsets(eq(processedOffsets)); File storeDir = testSideInputStorageManager.getStoreLocation(storeName); assertTrue("Store directory: " + storeDir.getPath() + " is missing.", storeDir.exists()); Map<SystemStreamPartition, String> fileOffsets = testSideInputStorageManager.getFileOffsets(); assertTrue("Failed to get offset for ssp: " + ssp.toString() + " from file.", fileOffsets.containsKey(ssp)); assertEquals("Mismatch between last processed offset and file offset.", fileOffsets.get(ssp), offset); }