TaskSideInputStorageManager { public Map<SystemStreamPartition, String> getFileOffsets() { LOG.info("Loading initial offsets from the file for side input stores."); Map<SystemStreamPartition, String> fileOffsets = new HashMap<>(); stores.keySet().forEach(storeName -> { LOG.debug("Reading local offsets for store: {}", storeName); File storeLocation = getStoreLocation(storeName); if (isValidSideInputStore(storeName, storeLocation)) { try { Map<SystemStreamPartition, String> offsets = storageManagerUtil.readOffsetFile(storeLocation, storeToSSps.get(storeName), true); fileOffsets.putAll(offsets); } catch (Exception e) { LOG.warn("Failed to load the offset file for side input store:" + storeName, e); } } }); return fileOffsets; } 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 testGetFileOffsets() { final String storeName = "test-get-file-offsets-store"; final String taskName = "test-get-file-offsets-task"; final String offset = "123"; Set<SystemStreamPartition> ssps = IntStream.range(1, 6) .mapToObj(idx -> new SystemStreamPartition("test-system", "test-stream", new Partition(idx))) .collect(Collectors.toSet()); TaskSideInputStorageManager testSideInputStorageManager = new MockTaskSideInputStorageManagerBuilder(taskName, LOGGED_STORE_DIR) .addLoggedStore(storeName, ssps) .build(); initializeSideInputStorageManager(testSideInputStorageManager); Map<SystemStreamPartition, String> processedOffsets = ssps.stream() .collect(Collectors.toMap(Function.identity(), ssp -> offset)); testSideInputStorageManager.writeFileOffsets(processedOffsets); Map<SystemStreamPartition, String> fileOffsets = testSideInputStorageManager.getFileOffsets(); ssps.forEach(ssp -> { 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); }); }