TaskSideInputStorageManager { public void stop(Map<SystemStreamPartition, String> lastProcessedOffsets) { LOG.info("Stopping the side input stores."); stores.values().forEach(StorageEngine::stop); 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 testStop() { final String storeName = "test-stop-store"; final String taskName = "test-stop-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, NON_LOGGED_STORE_DIR) .addInMemoryStore(storeName, ImmutableSet.of()) .build(); initializeSideInputStorageManager(testSideInputStorageManager); testSideInputStorageManager.stop(processedOffsets); verify(testSideInputStorageManager.getStore(storeName)).stop(); verify(testSideInputStorageManager).writeFileOffsets(eq(processedOffsets)); }