azkaban-aplcache

diff --git a/azkaban-common/src/main/java/azkaban/storage/HdfsStorage.java b/azkaban-common/src/main/java/azkaban/storage/HdfsStorage.java
index b0ef27e..3679150 100644
--- a/azkaban-common/src/main/java/azkaban/storage/HdfsStorage.java
+++ b/azkaban-common/src/main/java/azkaban/storage/HdfsStorage.java
@@ -60,7 +60,7 @@ public class HdfsStorage implements Storage {
   @Override
   public InputStream get(final String key) throws IOException {
     this.hdfsAuth.authorize();
-    return this.hdfs.open(new Path(this.rootUri.toString(), key));
+    return this.hdfs.open(fullPath(key));
   }
 
   @Override
@@ -102,6 +102,17 @@ public class HdfsStorage implements Storage {
 
   @Override
   public boolean delete(final String key) {
-    throw new UnsupportedOperationException("Method not implemented");
+    this.hdfsAuth.authorize();
+    final Path path = fullPath(key);
+    try {
+      return this.hdfs.delete(path, false);
+    } catch (final IOException e) {
+      log.error("HDFS delete failed on " + path, e);
+      return false;
+    }
+  }
+
+  private Path fullPath(final String key) {
+    return new Path(this.rootUri.toString(), key);
   }
 }

diff --git a/azkaban-common/src/test/java/azkaban/storage/HdfsStorageTest.java b/azkaban-common/src/test/java/azkaban/storage/HdfsStorageTest.java
index 99297c3..81218ef 100644
--- a/azkaban-common/src/test/java/azkaban/storage/HdfsStorageTest.java
+++ b/azkaban-common/src/test/java/azkaban/storage/HdfsStorageTest.java
@@ -74,4 +74,10 @@ public class HdfsStorageTest {
     final String expectedPath = "/path/to/foo/" + expectedName;
     verify(this.hdfs).copyFromLocalFile(new Path(file.getAbsolutePath()), new Path(expectedPath));
   }
+
+  @Test
+  public void testDelete() throws Exception {
+    this.hdfsStorage.delete("1/1-hash.zip");
+    verify(this.hdfs).delete(new Path("hdfs://localhost:9000/path/to/foo/1/1-hash.zip"), false);
+  }
 }