AsyncFSStorageAgent extends FSStorageAgent implements AsyncStorageAgent { @Override public void save(final Object object, final int operatorId, final long windowId) throws IOException { if (syncCheckpoint) { super.save(object, operatorId, windowId); return; } if (localBasePath == null) { synchronized (this) { if (localBasePath == null) { localBasePath = Files.createTempDirectory("chkp").toString(); logger.info("using {} as the basepath for checkpointing.", localBasePath); } } } String operatorIdStr = String.valueOf(operatorId); File directory = new File(localBasePath, operatorIdStr); if (!directory.exists()) { directory.mkdirs(); } try (FileOutputStream stream = new FileOutputStream(new File(directory, String.valueOf(windowId)))) { store(stream, object); } } @SuppressWarnings("unused") private  AsyncFSStorageAgent();  AsyncFSStorageAgent(String path, Configuration conf); @Deprecated  AsyncFSStorageAgent(String localBasePath, String path, Configuration conf); @Override void save(final Object object, final int operatorId, final long windowId); void copyToHDFS(final int operatorId, final long windowId); @Override void flush(int operatorId, long windowId); @Override Object readResolve(); @Override boolean isSyncCheckpoint(); void setSyncCheckpoint(boolean syncCheckpoint);  }
@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); testMeta.storageAgent.copyToHDFS(1, 1); @SuppressWarnings("unchecked") Map<Integer, String> decoded = (Map<Integer, String>)testMeta.storageAgent.load(1, 1); Assert.assertEquals("dataOf1", data, decoded); }