@Test public void testWriteWithGPGSymmetricEncryption() throws Exception { byte[] streamString = "testEncryptedContents".getBytes("UTF-8"); FileStatus status = fs.getFileStatus(testTempPath); OwnerAndPermission ownerAndPermission = new OwnerAndPermission(status.getOwner(), status.getGroup(), new FsPermission(FsAction.ALL, FsAction.ALL, FsAction.ALL)); CopyableFile cf = CopyableFileUtils.getTestCopyableFile(ownerAndPermission); CopyableDatasetMetadata metadata = new CopyableDatasetMetadata(new TestCopyableDataset(new Path("/source"))); WorkUnitState state = TestUtils.createTestWorkUnitState(); state.setProp(ConfigurationKeys.WRITER_STAGING_DIR, new Path(testTempPath, "staging").toString()); state.setProp(ConfigurationKeys.WRITER_OUTPUT_DIR, new Path(testTempPath, "output").toString()); state.setProp(ConfigurationKeys.WRITER_FILE_PATH, RandomStringUtils.randomAlphabetic(5)); state.setProp("writer.encrypt." + EncryptionConfigParser.ENCRYPTION_ALGORITHM_KEY, "gpg"); state.setProp("writer.encrypt." + EncryptionConfigParser.ENCRYPTION_KEYSTORE_PASSWORD_KEY, "testPassword"); CopySource.serializeCopyEntity(state, cf); CopySource.serializeCopyableDataset(state, metadata); FileAwareInputStreamDataWriter dataWriter = new FileAwareInputStreamDataWriter(state, 1, 0); FileAwareInputStream fileAwareInputStream = FileAwareInputStream.builder().file(cf) .inputStream(StreamUtils.convertStream(new ByteArrayInputStream(streamString))).build(); dataWriter.write(fileAwareInputStream); dataWriter.commit(); Path writtenFilePath = new Path(new Path(state.getProp(ConfigurationKeys.WRITER_OUTPUT_DIR), cf.getDatasetAndPartition(metadata).identifier()), cf.getDestination()); Assert.assertTrue(writtenFilePath.getName().endsWith("gpg"), "Expected encryption name to be appended to destination"); byte[] encryptedContent = IOUtils.toByteArray(new FileInputStream(writtenFilePath.toString())); byte[] decryptedContent = new byte[streamString.length]; IOUtils.readFully(GPGFileDecryptor.decryptFile(new FileInputStream(writtenFilePath.toString()), "testPassword"), decryptedContent); Assert.assertNotEquals(encryptedContent, streamString); Assert.assertEquals(decryptedContent, streamString); }
@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; }