FSStorageAgent implements StorageAgent, Serializable { @Override public Object load(int operatorId, long windowId) throws IOException { Path lPath = new Path(path + Path.SEPARATOR + String.valueOf(operatorId) + Path.SEPARATOR + Long.toHexString(windowId)); logger.debug("Loading: {}", lPath); FSDataInputStream stream = fileContext.open(lPath); try { return retrieve(stream); } finally { stream.close(); } } 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 testLoad() throws IOException { 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"); testMeta.storageAgent.save(dataOf1, 1, 1); testMeta.storageAgent.save(dataOf2, 2, 1); @SuppressWarnings("unchecked") Map<Integer, String> decoded1 = (Map<Integer, String>)testMeta.storageAgent.load(1, 1); @SuppressWarnings("unchecked") Map<Integer, String> decoded2 = (Map<Integer, String>)testMeta.storageAgent.load(2, 1); Assert.assertEquals("data of 1", dataOf1, decoded1); Assert.assertEquals("data of 2", dataOf2, decoded2); }