@Test public void testCommit() throws IOException { String destinationExistingToken = "destination"; String destinationAdditionalTokens = "path"; String fileName = "file"; Path destination = new Path(new Path(new Path("/", destinationExistingToken), destinationAdditionalTokens), fileName); Path destinationWithoutLeadingSeparator = new Path(new Path(destinationExistingToken, destinationAdditionalTokens), fileName); File tmpFile = Files.createTempDir(); tmpFile.deleteOnExit(); Path tmpPath = new Path(tmpFile.getAbsolutePath()); Path originFile = new Path(tmpPath, fileName); this.fs.createNewFile(originFile); Path stagingDir = new Path(tmpPath, "staging"); this.fs.mkdirs(stagingDir); Path outputDir = new Path(tmpPath, "output"); this.fs.mkdirs(outputDir); FileStatus status = this.fs.getFileStatus(originFile); FsPermission readWrite = new FsPermission(FsAction.READ_WRITE, FsAction.READ_WRITE, FsAction.READ_WRITE); FsPermission dirReadWrite = new FsPermission(FsAction.ALL, FsAction.READ_WRITE, FsAction.READ_WRITE); OwnerAndPermission ownerAndPermission = new OwnerAndPermission(status.getOwner(), status.getGroup(), readWrite); List<OwnerAndPermission> ancestorOwnerAndPermissions = Lists.newArrayList(); ancestorOwnerAndPermissions.add(ownerAndPermission); ancestorOwnerAndPermissions.add(ownerAndPermission); ancestorOwnerAndPermissions.add(ownerAndPermission); ancestorOwnerAndPermissions.add(ownerAndPermission); Properties properties = new Properties(); properties.setProperty(ConfigurationKeys.DATA_PUBLISHER_FINAL_DIR, "/publisher"); CopyableFile cf = CopyableFile.fromOriginAndDestination(this.fs, status, destination, CopyConfiguration.builder(FileSystem.getLocal(new Configuration()), properties).publishDir(new Path("/target")) .preserve(PreserveAttributes.fromMnemonicString("")).build()) .destinationOwnerAndPermission(ownerAndPermission) .ancestorsOwnerAndPermission(ancestorOwnerAndPermissions) .build(); WorkUnitState state = TestUtils.createTestWorkUnitState(); state.setProp(ConfigurationKeys.WRITER_STAGING_DIR, stagingDir.toUri().getPath()); state.setProp(ConfigurationKeys.WRITER_OUTPUT_DIR, outputDir.toUri().getPath()); state.setProp(ConfigurationKeys.WRITER_FILE_PATH, RandomStringUtils.randomAlphabetic(5)); CopyableDatasetMetadata metadata = new CopyableDatasetMetadata(new TestCopyableDataset(new Path("/source"))); CopySource.serializeCopyEntity(state, cf); CopySource.serializeCopyableDataset(state, metadata); FileAwareInputStreamDataWriter writer = new FileAwareInputStreamDataWriter(state, 1, 0); Path writtenFile = writer.getStagingFilePath(cf); this.fs.mkdirs(writtenFile.getParent()); this.fs.createNewFile(writtenFile); Path outputRoot = FileAwareInputStreamDataWriter.getPartitionOutputRoot(outputDir, cf.getDatasetAndPartition(metadata)); Path existingOutputPath = new Path(outputRoot, destinationExistingToken); this.fs.mkdirs(existingOutputPath); FileStatus fileStatus = this.fs.getFileStatus(existingOutputPath); FsPermission existingPathPermission = fileStatus.getPermission(); Assert.assertTrue(this.fs.exists(existingOutputPath)); Assert.assertEquals(this.fs.listStatus(existingOutputPath).length, 0); writer.actualProcessedCopyableFile = Optional.of(cf); writer.commit(); Path expectedOutputPath = new Path(outputRoot, destinationWithoutLeadingSeparator); Assert.assertTrue(this.fs.exists(expectedOutputPath)); fileStatus = this.fs.getFileStatus(expectedOutputPath); Assert.assertEquals(fileStatus.getOwner(), ownerAndPermission.getOwner()); Assert.assertEquals(fileStatus.getGroup(), ownerAndPermission.getGroup()); Assert.assertEquals(fileStatus.getPermission(), readWrite); fileStatus = this.fs.getFileStatus(expectedOutputPath.getParent()); Assert.assertEquals(fileStatus.getPermission(), dirReadWrite); fileStatus = this.fs.getFileStatus(existingOutputPath); Assert.assertEquals(fileStatus.getPermission(), existingPathPermission); Assert.assertFalse(this.fs.exists(writer.stagingDir)); }
@Override public void commit() throws IOException { if (!this.actualProcessedCopyableFile.isPresent()) { return; } CopyableFile copyableFile = this.actualProcessedCopyableFile.get(); Path stagingFilePath = getStagingFilePath(copyableFile); Path outputFilePath = getSplitOutputFilePath(copyableFile, this.outputDir, copyableFile.getDatasetAndPartition(this.copyableDatasetMetadata), this.state); log.info(String.format("Committing data from %s to %s", stagingFilePath, outputFilePath)); try { setFilePermissions(copyableFile); Iterator<OwnerAndPermission> ancestorOwnerAndPermissionIt = copyableFile.getAncestorsOwnerAndPermission() == null ? Iterators.<OwnerAndPermission>emptyIterator() : copyableFile.getAncestorsOwnerAndPermission().iterator(); ensureDirectoryExists(this.fs, outputFilePath.getParent(), ancestorOwnerAndPermissionIt); this.fileContext.rename(stagingFilePath, outputFilePath, renameOptions); } catch (IOException ioe) { log.error("Could not commit file %s.", outputFilePath); this.recoveryHelper.persistFile(this.state, copyableFile, stagingFilePath); throw ioe; } finally { try { this.fs.delete(this.stagingDir, true); } catch (IOException ioe) { log.warn("Failed to delete staging path at " + this.stagingDir); } } }
FileAwareInputStreamDataWriter extends InstrumentedDataWriter<FileAwareInputStream> implements FinalState, SpeculativeAttemptAwareConstruct { @Override public void commit() throws IOException { if (!this.actualProcessedCopyableFile.isPresent()) { return; } CopyableFile copyableFile = this.actualProcessedCopyableFile.get(); Path stagingFilePath = getStagingFilePath(copyableFile); Path outputFilePath = getSplitOutputFilePath(copyableFile, this.outputDir, copyableFile.getDatasetAndPartition(this.copyableDatasetMetadata), this.state); log.info(String.format("Committing data from %s to %s", stagingFilePath, outputFilePath)); try { setFilePermissions(copyableFile); Iterator<OwnerAndPermission> ancestorOwnerAndPermissionIt = copyableFile.getAncestorsOwnerAndPermission() == null ? Iterators.<OwnerAndPermission>emptyIterator() : copyableFile.getAncestorsOwnerAndPermission().iterator(); ensureDirectoryExists(this.fs, outputFilePath.getParent(), ancestorOwnerAndPermissionIt); this.fileContext.rename(stagingFilePath, outputFilePath, renameOptions); } catch (IOException ioe) { log.error("Could not commit file %s.", outputFilePath); this.recoveryHelper.persistFile(this.state, copyableFile, stagingFilePath); throw ioe; } finally { try { this.fs.delete(this.stagingDir, true); } catch (IOException ioe) { log.warn("Failed to delete staging path at " + this.stagingDir); } } } }
FileAwareInputStreamDataWriter extends InstrumentedDataWriter<FileAwareInputStream> implements FinalState, SpeculativeAttemptAwareConstruct { @Override public void commit() throws IOException { if (!this.actualProcessedCopyableFile.isPresent()) { return; } CopyableFile copyableFile = this.actualProcessedCopyableFile.get(); Path stagingFilePath = getStagingFilePath(copyableFile); Path outputFilePath = getSplitOutputFilePath(copyableFile, this.outputDir, copyableFile.getDatasetAndPartition(this.copyableDatasetMetadata), this.state); log.info(String.format("Committing data from %s to %s", stagingFilePath, outputFilePath)); try { setFilePermissions(copyableFile); Iterator<OwnerAndPermission> ancestorOwnerAndPermissionIt = copyableFile.getAncestorsOwnerAndPermission() == null ? Iterators.<OwnerAndPermission>emptyIterator() : copyableFile.getAncestorsOwnerAndPermission().iterator(); ensureDirectoryExists(this.fs, outputFilePath.getParent(), ancestorOwnerAndPermissionIt); this.fileContext.rename(stagingFilePath, outputFilePath, renameOptions); } catch (IOException ioe) { log.error("Could not commit file %s.", outputFilePath); this.recoveryHelper.persistFile(this.state, copyableFile, stagingFilePath); throw ioe; } finally { try { this.fs.delete(this.stagingDir, true); } catch (IOException ioe) { log.warn("Failed to delete staging path at " + this.stagingDir); } } } FileAwareInputStreamDataWriter(State state, int numBranches, int branchId, String writerAttemptId);  FileAwareInputStreamDataWriter(State state, int numBranches, int branchId); }
FileAwareInputStreamDataWriter extends InstrumentedDataWriter<FileAwareInputStream> implements FinalState, SpeculativeAttemptAwareConstruct { @Override public void commit() throws IOException { if (!this.actualProcessedCopyableFile.isPresent()) { return; } CopyableFile copyableFile = this.actualProcessedCopyableFile.get(); Path stagingFilePath = getStagingFilePath(copyableFile); Path outputFilePath = getSplitOutputFilePath(copyableFile, this.outputDir, copyableFile.getDatasetAndPartition(this.copyableDatasetMetadata), this.state); log.info(String.format("Committing data from %s to %s", stagingFilePath, outputFilePath)); try { setFilePermissions(copyableFile); Iterator<OwnerAndPermission> ancestorOwnerAndPermissionIt = copyableFile.getAncestorsOwnerAndPermission() == null ? Iterators.<OwnerAndPermission>emptyIterator() : copyableFile.getAncestorsOwnerAndPermission().iterator(); ensureDirectoryExists(this.fs, outputFilePath.getParent(), ancestorOwnerAndPermissionIt); this.fileContext.rename(stagingFilePath, outputFilePath, renameOptions); } catch (IOException ioe) { log.error("Could not commit file %s.", outputFilePath); this.recoveryHelper.persistFile(this.state, copyableFile, stagingFilePath); throw ioe; } finally { try { this.fs.delete(this.stagingDir, true); } catch (IOException ioe) { log.warn("Failed to delete staging path at " + this.stagingDir); } } } FileAwareInputStreamDataWriter(State state, int numBranches, int branchId, String writerAttemptId);  FileAwareInputStreamDataWriter(State state, int numBranches, int branchId); @Override final void writeImpl(FileAwareInputStream fileAwareInputStream); static Path getOutputFilePath(CopyableFile file, Path outputDir,
      CopyEntity.DatasetAndPartition datasetAndPartition); static Path getSplitOutputFilePath(CopyableFile file, Path outputDir,
      CopyableFile.DatasetAndPartition datasetAndPartition, State workUnit); static Path getOutputDir(State state); @Override long recordsWritten(); @Override long bytesWritten(); @Override void commit(); @Override void cleanup(); @Override State getFinalState(); @Override boolean isSpeculativeAttemptSafe(); }
FileAwareInputStreamDataWriter extends InstrumentedDataWriter<FileAwareInputStream> implements FinalState, SpeculativeAttemptAwareConstruct { @Override public void commit() throws IOException { if (!this.actualProcessedCopyableFile.isPresent()) { return; } CopyableFile copyableFile = this.actualProcessedCopyableFile.get(); Path stagingFilePath = getStagingFilePath(copyableFile); Path outputFilePath = getSplitOutputFilePath(copyableFile, this.outputDir, copyableFile.getDatasetAndPartition(this.copyableDatasetMetadata), this.state); log.info(String.format("Committing data from %s to %s", stagingFilePath, outputFilePath)); try { setFilePermissions(copyableFile); Iterator<OwnerAndPermission> ancestorOwnerAndPermissionIt = copyableFile.getAncestorsOwnerAndPermission() == null ? Iterators.<OwnerAndPermission>emptyIterator() : copyableFile.getAncestorsOwnerAndPermission().iterator(); ensureDirectoryExists(this.fs, outputFilePath.getParent(), ancestorOwnerAndPermissionIt); this.fileContext.rename(stagingFilePath, outputFilePath, renameOptions); } catch (IOException ioe) { log.error("Could not commit file %s.", outputFilePath); this.recoveryHelper.persistFile(this.state, copyableFile, stagingFilePath); throw ioe; } finally { try { this.fs.delete(this.stagingDir, true); } catch (IOException ioe) { log.warn("Failed to delete staging path at " + this.stagingDir); } } } FileAwareInputStreamDataWriter(State state, int numBranches, int branchId, String writerAttemptId);  FileAwareInputStreamDataWriter(State state, int numBranches, int branchId); @Override final void writeImpl(FileAwareInputStream fileAwareInputStream); static Path getOutputFilePath(CopyableFile file, Path outputDir,
      CopyEntity.DatasetAndPartition datasetAndPartition); static Path getSplitOutputFilePath(CopyableFile file, Path outputDir,
      CopyableFile.DatasetAndPartition datasetAndPartition, State workUnit); static Path getOutputDir(State state); @Override long recordsWritten(); @Override long bytesWritten(); @Override void commit(); @Override void cleanup(); @Override State getFinalState(); @Override boolean isSpeculativeAttemptSafe(); static final String GOBBLIN_COPY_BYTES_COPIED_METER; static final String GOBBLIN_COPY_CHECK_FILESIZE; static final boolean DEFAULT_GOBBLIN_COPY_CHECK_FILESIZE; static final String GOBBLIN_COPY_TASK_OVERWRITE_ON_COMMIT; static final boolean DEFAULT_GOBBLIN_COPY_TASK_OVERWRITE_ON_COMMIT; static final String STAGING_DIR_SUFFIX; static final String DATASET_STAGING_DIR_PATH; }