FSWindowDataManager implements WindowDataManager { @Override public Map<Integer, Object> retrieveAllPartitions(long windowId) throws IOException { if (windowId > largestCompletedWindow) { return null; } Map<Integer, Object> artifacts = Maps.newHashMap(); Object artifact = retrieve(wal, windowId); if (artifact != null) { artifacts.put(operatorId, artifact); } if (repartitioned) { for (Map.Entry<Integer, FSWindowReplayWAL> entry : readOnlyWals.entrySet()) { artifact = retrieve(entry.getValue(), windowId); if (artifact != null) { artifacts.put(entry.getKey(), artifact); } } } return artifacts; } 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 testRetrieveAllPartitions() 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.teardown(); pair2.second.teardown(); List<WindowDataManager> managers = pair1.second.partition(3, null); managers.get(0).setup(pair1.first); Map<Integer, Object> artifacts = managers.get(0).retrieveAllPartitions(1); Assert.assertEquals("num artifacts", 2, artifacts.size()); Assert.assertEquals("artifact 1", dataOf1, artifacts.get(1)); Assert.assertEquals("artifact 2", dataOf2, artifacts.get(2)); managers.get(0).teardown(); }