From ea408f531eac216c6de4bd393f6a3f249ac46797 Mon Sep 17 00:00:00 2001
From: Himanshu Gupta <g.himanshu@gmail.com>
Date: Mon, 7 Mar 2016 16:51:41 -0600
Subject: [PATCH] let the hdfs segment kill be success when segment file does
 not exist

---
 .../storage/hdfs/HdfsDataSegmentKiller.java   | 25 ++++++-------------
 .../hdfs/HdfsDataSegmentKillerTest.java       |  8 ++++++
 2 files changed, 15 insertions(+), 18 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 d2262d0ccb3..bb9d3a000f2 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
@@ -50,10 +50,16 @@ public class HdfsDataSegmentKiller implements DataSegmentKiller
     final Path path = getPath(segment);
     log.info("killing segment[%s] mapped to path[%s]", segment.getIdentifier(), path);
 
-    final FileSystem fs = checkPathAndGetFilesystem(path);
     try {
       if (path.getName().endsWith(".zip")) {
 
+        final FileSystem fs = path.getFileSystem(config);
+
+        if (!fs.exists(path)) {
+          log.warn("Segment Path [%s] does not exist. It appears to have been deleted already.", path);
+          return ;
+        }
+
         // path format -- > .../dataSource/interval/version/partitionNum/xxx.zip
         Path partitionNumDir = path.getParent();
         if (!fs.delete(partitionNumDir, true)) {
@@ -95,21 +101,4 @@ public class HdfsDataSegmentKiller implements DataSegmentKiller
   {
     return new Path(String.valueOf(segment.getLoadSpec().get(PATH_KEY)));
   }
-
-  private FileSystem checkPathAndGetFilesystem(Path path) throws SegmentLoadingException
-  {
-    FileSystem fs;
-    try {
-      fs = path.getFileSystem(config);
-
-      if (!fs.exists(path)) {
-        throw new SegmentLoadingException("Path[%s] doesn't exist.", path);
-      }
-
-      return fs;
-    }
-    catch (IOException e) {
-      throw new SegmentLoadingException(e, "Problems interacting with filesystem[%s].", path);
-    }
-  }
 }
diff --git a/extensions/hdfs-storage/src/test/java/io/druid/storage/hdfs/HdfsDataSegmentKillerTest.java b/extensions/hdfs-storage/src/test/java/io/druid/storage/hdfs/HdfsDataSegmentKillerTest.java
index 53d6bbd33e3..3496a71d01c 100644
--- a/extensions/hdfs-storage/src/test/java/io/druid/storage/hdfs/HdfsDataSegmentKillerTest.java
+++ b/extensions/hdfs-storage/src/test/java/io/druid/storage/hdfs/HdfsDataSegmentKillerTest.java
@@ -95,6 +95,14 @@ public class HdfsDataSegmentKillerTest
     Assert.assertFalse(fs.exists(dataSourceDir));
   }
 
+  @Test
+  public void testKillNonExistingSegment() throws Exception
+  {
+    Configuration config = new Configuration();
+    HdfsDataSegmentKiller killer = new HdfsDataSegmentKiller(config);
+    killer.kill(getSegmentWithPath(new Path("/xxx/", "index.zip").toString()));
+  }
+
   private void makePartitionDirWithIndex(FileSystem fs, Path path) throws IOException
   {
     Assert.assertTrue(fs.mkdirs(path));