HdfsStorage implements Storage { @VisibleForTesting static void uploadLocalFile(final Path localPath, final Path remotePath, final FileContext remoteFileContext) throws IOException { FSDataInputStream localStream = null; FSDataOutputStream remoteStream = null; final String actionString = String.format("upload of local file %s to remote file %s", localPath, remotePath); log.debug("Starting " + actionString); try { final FileContext localFileContext = FileContext.getLocalFSFileContext(); localStream = localFileContext.open(localPath); remoteStream = remoteFileContext.create(remotePath, EnumSet.of(CreateFlag.CREATE, CreateFlag.OVERWRITE)); IOUtils.copyBytes(localStream, remoteStream, UPLOAD_BUFFER_SIZE_BYTES, true); } catch (final IOException e) { log.error("Error during " + actionString, e); throw e; } finally { closeStreamsQuietly(localStream, remoteStream); log.debug("Ending " + actionString); } } @Inject  HdfsStorage(final AzkabanCommonModuleConfig config, final HdfsAuth hdfsAuth,
      @Named("hdfsFileContext") final FileContext hdfsFileContext,
      @Named("hdfs_cached_httpFS") @Nullable final FileSystem http); @Override InputStream getProject(final String key); @Override String putProject(final ProjectStorageMetadata metadata, final File localFile); @Override InputStream getDependency(final Dependency dep); @Override boolean dependencyFetchingEnabled(); @Override boolean deleteProject(final String key); @Override String getDependencyRootPath();  }
@Test public void testUploadFile() throws Exception{ final File sourceFile = new File( getClass().getClassLoader().getResource("sample_flow_01.zip").getFile()); final String sourceHash = new String(Hex.encodeHex(HashUtils.MD5.getHashBytes(sourceFile))); final File tmpUploadDir = TEMP_DIR.newFolder("upload_test_dir"); final File destFile = new File(tmpUploadDir, "tmpfile1"); final FileContext destFileContext = FileContext.getLocalFSFileContext(); HdfsStorage.uploadLocalFile(new Path(sourceFile.getPath()), new Path(destFile.getPath()), destFileContext); final String destHash = new String(Hex.encodeHex(HashUtils.MD5.getHashBytes(destFile))); Assert.assertEquals(sourceHash, destHash); final byte[] newFileContent = "12345".getBytes(UTF_8); final File newSourceFile = new File(tmpUploadDir, "tmpsource1"); FileUtils.writeByteArrayToFile(newSourceFile, newFileContent, false); HdfsStorage.uploadLocalFile(new Path(newSourceFile.getPath()), new Path(destFile.getPath()), destFileContext); final byte[] destFileContent = FileUtils.readFileToByteArray(destFile); FileUtils.deleteDirectory(tmpUploadDir); Assert.assertArrayEquals(newFileContent, destFileContent); }