HdfsStorage implements Storage { @Override public String putProject(final ProjectStorageMetadata metadata, final File localFile) { this.hdfsAuth.authorize(); final Path projectsPath = new Path(this.projectRootUri.getPath(), String.valueOf(metadata.getProjectId())); try { this.hdfsFileContext.mkdir(projectsPath, FsPermission.getDefault(), true); log.info("Created project dir: " + projectsPath); final Path targetPath = new Path(projectsPath, StorageUtils.getTargetProjectFilename(metadata.getProjectId(), metadata.getHash())); final Path tmpPath = new Path(projectsPath, TMP_PROJECT_UPLOAD_FILENAME); final Path localFilePath = new Path(localFile.getPath()) .makeQualified(FsConstants.LOCAL_FS_URI, null); log.info("Scheme qualified path of the local zip file is " + localFilePath); log.info(String.format("Creating project artifact: meta: %s path: %s", metadata, targetPath)); uploadLocalFile(localFilePath, tmpPath, hdfsFileContext); this.hdfsFileContext.rename(tmpPath, targetPath, Options.Rename.OVERWRITE); return getRelativeProjectPath(targetPath); } catch (final IOException e) { log.error("error in putProject(): Metadata: " + metadata); throw new StorageException(e); } } @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 testPutProject() throws Exception { final File file = new File( getClass().getClassLoader().getResource("sample_flow_01.zip").getFile()); final String hash = new String(Hex.encodeHex(HashUtils.MD5.getHashBytes(file))); FSDataOutputStream nullOutputStream = new FSDataOutputStream(new NullOutputStream()); when(this.hdfsFileContext.create(any(Path.class), any(EnumSet.class))) .thenReturn(nullOutputStream); final ProjectStorageMetadata metadata = new ProjectStorageMetadata(1, 2, "uploader", HashUtils.MD5.getHashBytes(file), IPv4); final String key = this.hdfsStorage.putProject(metadata, file); final String expectedName = String.format("1/1-%s.zip", hash); Assert.assertEquals(expectedName, key); final String expectedPath = "/path/to/prj/" + expectedName; verify(this.hdfsFileContext).create(any(Path.class), eq(EnumSet.of(CreateFlag.CREATE, CreateFlag.OVERWRITE))); verify(this.hdfsFileContext) .rename(any(Path.class), eq(new Path(expectedPath)), eq(Options.Rename.OVERWRITE)); }