From c2bebfe39e9d4507edb5560a93a0def262e39ea9 Mon Sep 17 00:00:00 2001
From: Himanshu Gupta <g.himanshu@gmail.com>
Date: Sun, 23 Aug 2015 22:06:12 -0500
Subject: [PATCH] delete version, interval, dataSource directories on segment
 deletion if possible, so that they are not left behind and consume ns quota
 on HDFS

---
 .../storage/hdfs/HdfsDataSegmentKiller.java   | 31 +++++++++++++++++--
 1 file changed, 29 insertions(+), 2 deletions(-)

diff --git a/extensions/hdfs-storage/src/main/java/io/druid/storage/hdfs/HdfsDataSegmentKiller.java b/extensions/hdfs-storage/src/main/java/io/druid/storage/hdfs/HdfsDataSegmentKiller.java
index 8a77ad85f95..d1314162252 100644
--- a/extensions/hdfs-storage/src/main/java/io/druid/storage/hdfs/HdfsDataSegmentKiller.java
+++ b/extensions/hdfs-storage/src/main/java/io/druid/storage/hdfs/HdfsDataSegmentKiller.java
@@ -44,8 +44,25 @@ public class HdfsDataSegmentKiller implements DataSegmentKiller
     final FileSystem fs = checkPathAndGetFilesystem(path);
     try {
       if (path.getName().endsWith(".zip")) {
-        // delete the parent directory containing the zip file and the descriptor
-        fs.delete(path.getParent(), true);
+
+        // path format -- > .../dataSource/interval/version/partitionNum/xxx.zip
+        Path partitionNumDir = path.getParent();
+        if (!fs.delete(partitionNumDir, true)) {
+          throw new SegmentLoadingException(
+              "Unable to kill segment, failed to delete dir [%s]",
+              partitionNumDir.toString()
+          );
+        }
+
+        //try to delete other directories if possible
+        Path versionDir = partitionNumDir.getParent();
+        if (safeNonRecursiveDelete(fs, versionDir)) {
+          Path intervalDir = versionDir.getParent();
+          if (safeNonRecursiveDelete(fs, intervalDir)) {
+            Path dataSourceDir = intervalDir.getParent();
+            safeNonRecursiveDelete(fs, dataSourceDir);
+          }
+        }
       } else {
         throw new SegmentLoadingException("Unknown file type[%s]", path);
       }
@@ -55,6 +72,16 @@ public class HdfsDataSegmentKiller implements DataSegmentKiller
     }
   }
 
+  private boolean safeNonRecursiveDelete(FileSystem fs, Path path)
+  {
+    try {
+      return fs.delete(path, false);
+    }
+    catch (Exception ex) {
+      return false;
+    }
+  }
+
   private Path getPath(DataSegment segment)
   {
     return new Path(String.valueOf(segment.getLoadSpec().get("path")));