FSWindowDataManager implements WindowDataManager { @Override public Object retrieve(long windowId) throws IOException { return retrieve(wal, windowId); } FSWindowDataManager(); @Override void setup(Context.OperatorContext context); @Override void save(Object object, long windowId); @Override Object retrieve(long windowId); @Override Map<Integer, Object> retrieveAllPartitions(long windowId); @Override void committed(long committedWindowId); long getLargestCompletedWindow(); @Override List<WindowDataManager> partition(int newCount, Set<Integer> removedOperatorIds); @Override void teardown(); @VisibleForTesting Set<Integer> getDeletedOperators(); String getStatePath(); void setStatePath(String statePath); boolean isStatePathRelativeToAppPath(); void setStatePathRelativeToAppPath(boolean statePathRelativeToAppPath);  }
@Test public void testRetrieve() throws IOException { Pair<Context.OperatorContext, FSWindowDataManager> pair1 = createManagerAndContextFor(1); Pair<Context.OperatorContext, FSWindowDataManager> pair2 = createManagerAndContextFor(2); pair1.second.setup(pair1.first); pair2.second.setup(pair2.first); Map<Integer, String> dataOf1 = Maps.newHashMap(); dataOf1.put(1, "one"); dataOf1.put(2, "two"); dataOf1.put(3, "three"); Map<Integer, String> dataOf2 = Maps.newHashMap(); dataOf2.put(4, "four"); dataOf2.put(5, "five"); dataOf2.put(6, "six"); pair1.second.save(dataOf1, 1); pair2.second.save(dataOf2, 1); pair1.second.setup(pair1.first); Object artifact1 = pair1.second.retrieve(1); Assert.assertEquals("data of 1", dataOf1, artifact1); pair2.second.setup(pair2.first); Object artifact2 = pair2.second.retrieve(1); Assert.assertEquals("data of 2", dataOf2, artifact2); pair1.second.teardown(); pair2.second.teardown(); }