FSStorageAgent implements StorageAgent, Serializable { @SuppressWarnings("ThrowFromFinallyBlock") @Override public void save(Object object, int operatorId, long windowId) throws IOException { String operatorIdStr = String.valueOf(operatorId); Path lPath = new Path(path + Path.SEPARATOR + operatorIdStr + Path.SEPARATOR + TMP_FILE); String window = Long.toHexString(windowId); boolean stateSaved = false; FSDataOutputStream stream = null; try { stream = fileContext.create(lPath, EnumSet.of(CreateFlag.CREATE, CreateFlag.OVERWRITE), Options.CreateOpts.CreateParent.createParent()); store(stream, object); stateSaved = true; } catch (Throwable t) { logger.debug("while saving {} {}", operatorId, window, t); stateSaved = false; throw Throwables.propagate(t); } finally { try { if (stream != null) { stream.close(); } } catch (IOException ie) { stateSaved = false; throw new RuntimeException(ie); } finally { if (stateSaved) { logger.debug("Saving {}: {}", operatorId, window); fileContext.rename(lPath, new Path(path + Path.SEPARATOR + operatorIdStr + Path.SEPARATOR + window), Options.Rename.OVERWRITE); } } } } protected  FSStorageAgent();  FSStorageAgent(String path, Configuration conf); @SuppressWarnings("ThrowFromFinallyBlock") @Override void save(Object object, int operatorId, long windowId); @Override Object load(int operatorId, long windowId); @Override void delete(int operatorId, long windowId); @Override long[] getWindowIds(int operatorId); static void store(OutputStream stream, Object operator); static Object retrieve(InputStream stream); Object readResolve(); static final String TMP_FILE; final String path; }
@Test public void testSave() throws IOException { Map<Integer, String> data = Maps.newHashMap(); data.put(1, "one"); data.put(2, "two"); data.put(3, "three"); testMeta.storageAgent.save(data, 1, 1); @SuppressWarnings("unchecked") Map<Integer, String> decoded = (Map<Integer, String>)testMeta.storageAgent.load(1, 1); Assert.assertEquals("dataOf1", data, decoded); }