@Test public void testPersistFile() throws Exception { String content = "contents"; File stagingDir = Files.createTempDir(); stagingDir.deleteOnExit(); File file = new File(stagingDir, "file"); OutputStream os = new FileOutputStream(file); IOUtils.write(content, os); os.close(); Assert.assertEquals(stagingDir.listFiles().length, 1); State state = new State(); state.setProp(RecoveryHelper.PERSIST_DIR_KEY, this.tmpDir.getAbsolutePath()); state.setProp(ConfigurationKeys.DATA_PUBLISHER_FINAL_DIR, "/publisher"); File recoveryDir = new File(RecoveryHelper.getPersistDir(state).get().toUri().getPath()); FileSystem fs = FileSystem.getLocal(new Configuration()); CopyableFile copyableFile = CopyableFile.builder(fs, new FileStatus(0, false, 0, 0, 0, new Path("/file")), new Path("/dataset"), CopyConfiguration.builder(fs, state.getProperties()).preserve(PreserveAttributes.fromMnemonicString("")).build()).build(); CopySource.setWorkUnitGuid(state, Guid.fromHasGuid(copyableFile)); RecoveryHelper recoveryHelper = new RecoveryHelper(FileSystem.getLocal(new Configuration()), state); recoveryHelper.persistFile(state, copyableFile, new Path(file.getAbsolutePath())); Assert.assertEquals(stagingDir.listFiles().length, 0); Assert.assertEquals(recoveryDir.listFiles().length, 1); File fileInRecovery = recoveryDir.listFiles()[0].listFiles()[0]; Assert.assertEquals(IOUtils.readLines(new FileInputStream(fileInRecovery)).get(0), content); Optional<FileStatus> fileToRecover = recoveryHelper.findPersistedFile(state, copyableFile, Predicates.<FileStatus>alwaysTrue()); Assert.assertTrue(fileToRecover.isPresent()); Assert.assertEquals(fileToRecover.get().getPath().toUri().getPath(), fileInRecovery.getAbsolutePath()); fileToRecover = recoveryHelper.findPersistedFile(state, copyableFile, Predicates.<FileStatus>alwaysFalse()); Assert.assertFalse(fileToRecover.isPresent()); }
public boolean persistFile(State state, CopyableFile file, Path path) throws IOException { if (!this.persistDir.isPresent()) { return false; } String guid = computeGuid(state, file); Path guidPath = new Path(this.persistDir.get(), guid); if (!this.fs.exists(guidPath)) { this.fs.mkdirs(guidPath, new FsPermission(FsAction.ALL, FsAction.READ, FsAction.NONE)); } Path targetPath = new Path(guidPath, shortenPathName(file.getOrigin().getPath(), 250 - guid.length())); log.info(String.format("Persisting file %s with guid %s to location %s.", path, guid, targetPath)); if (this.fs.rename(path, targetPath)) { this.fs.setTimes(targetPath, System.currentTimeMillis(), -1); return true; } return false; }
RecoveryHelper { public boolean persistFile(State state, CopyableFile file, Path path) throws IOException { if (!this.persistDir.isPresent()) { return false; } String guid = computeGuid(state, file); Path guidPath = new Path(this.persistDir.get(), guid); if (!this.fs.exists(guidPath)) { this.fs.mkdirs(guidPath, new FsPermission(FsAction.ALL, FsAction.READ, FsAction.NONE)); } Path targetPath = new Path(guidPath, shortenPathName(file.getOrigin().getPath(), 250 - guid.length())); log.info(String.format("Persisting file %s with guid %s to location %s.", path, guid, targetPath)); if (this.fs.rename(path, targetPath)) { this.fs.setTimes(targetPath, System.currentTimeMillis(), -1); return true; } return false; } }
RecoveryHelper { public boolean persistFile(State state, CopyableFile file, Path path) throws IOException { if (!this.persistDir.isPresent()) { return false; } String guid = computeGuid(state, file); Path guidPath = new Path(this.persistDir.get(), guid); if (!this.fs.exists(guidPath)) { this.fs.mkdirs(guidPath, new FsPermission(FsAction.ALL, FsAction.READ, FsAction.NONE)); } Path targetPath = new Path(guidPath, shortenPathName(file.getOrigin().getPath(), 250 - guid.length())); log.info(String.format("Persisting file %s with guid %s to location %s.", path, guid, targetPath)); if (this.fs.rename(path, targetPath)) { this.fs.setTimes(targetPath, System.currentTimeMillis(), -1); return true; } return false; } RecoveryHelper(FileSystem fs, State state); }
RecoveryHelper { public boolean persistFile(State state, CopyableFile file, Path path) throws IOException { if (!this.persistDir.isPresent()) { return false; } String guid = computeGuid(state, file); Path guidPath = new Path(this.persistDir.get(), guid); if (!this.fs.exists(guidPath)) { this.fs.mkdirs(guidPath, new FsPermission(FsAction.ALL, FsAction.READ, FsAction.NONE)); } Path targetPath = new Path(guidPath, shortenPathName(file.getOrigin().getPath(), 250 - guid.length())); log.info(String.format("Persisting file %s with guid %s to location %s.", path, guid, targetPath)); if (this.fs.rename(path, targetPath)) { this.fs.setTimes(targetPath, System.currentTimeMillis(), -1); return true; } return false; } RecoveryHelper(FileSystem fs, State state); static Optional<Path> getPersistDir(State state); boolean persistFile(State state, CopyableFile file, Path path); Optional<FileStatus> findPersistedFile(State state, CopyEntity file, Predicate<FileStatus> filter); void purgeOldPersistedFile(); }
RecoveryHelper { public boolean persistFile(State state, CopyableFile file, Path path) throws IOException { if (!this.persistDir.isPresent()) { return false; } String guid = computeGuid(state, file); Path guidPath = new Path(this.persistDir.get(), guid); if (!this.fs.exists(guidPath)) { this.fs.mkdirs(guidPath, new FsPermission(FsAction.ALL, FsAction.READ, FsAction.NONE)); } Path targetPath = new Path(guidPath, shortenPathName(file.getOrigin().getPath(), 250 - guid.length())); log.info(String.format("Persisting file %s with guid %s to location %s.", path, guid, targetPath)); if (this.fs.rename(path, targetPath)) { this.fs.setTimes(targetPath, System.currentTimeMillis(), -1); return true; } return false; } RecoveryHelper(FileSystem fs, State state); static Optional<Path> getPersistDir(State state); boolean persistFile(State state, CopyableFile file, Path path); Optional<FileStatus> findPersistedFile(State state, CopyEntity file, Predicate<FileStatus> filter); void purgeOldPersistedFile(); static final String PERSIST_DIR_KEY; static final String PERSIST_RETENTION_KEY; static final int DEFAULT_PERSIST_RETENTION; }