FSWindowDataManager implements WindowDataManager { @Override public void save(Object object, long windowId) throws IOException { closeReaders(); FileSystemWAL.FileSystemWALWriter writer = wal.getWriter(); byte[] windowIdBytes = Longs.toByteArray(windowId); writer.append(new Slice(windowIdBytes)); writer.append(toSlice(object)); serializationBuffer.reset(); wal.beforeCheckpoint(windowId); wal.windowWalParts.put(windowId, writer.getCurrentPointer().getPartNum()); writer.rotateIfNecessary(); } 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 testSave() throws IOException { Pair<Context.OperatorContext, FSWindowDataManager> pair = createManagerAndContextFor(1); pair.second.setup(pair.first); Map<Integer, String> data = Maps.newHashMap(); data.put(1, "one"); data.put(2, "two"); data.put(3, "three"); pair.second.save(data, 1); pair.second.setup(pair.first); @SuppressWarnings("unchecked") Map<Integer, String> artifact = (Map<Integer, String>)pair.second.retrieve(1); Assert.assertEquals("dataOf1", data, artifact); pair.second.teardown(); }