some minor cleanups to ingest firehose

indexing-service/src/main/java/io/druid/indexing/common/task/Task.java

@@ -24,7 +24,7 @@ import com.fasterxml.jackson.annotation.JsonTypeInfo;
 import io.druid.indexing.common.TaskStatus;
 import io.druid.indexing.common.TaskToolbox;
 import io.druid.indexing.common.actions.TaskActionClient;
-import io.druid.indexing.firehose.IngestTask;
+import io.druid.indexing.firehose.ReIngestTask;
 import io.druid.query.Query;
 import io.druid.query.QueryRunner;
 
@@ -55,7 +55,7 @@ import io.druid.query.QueryRunner;
     @JsonSubTypes.Type(name = "noop", value = NoopTask.class),
     @JsonSubTypes.Type(name = "version_converter", value = VersionConverterTask.class),
     @JsonSubTypes.Type(name = "version_converter_sub", value = VersionConverterTask.SubTask.class),
-    @JsonSubTypes.Type(name = "ingest-task", value = IngestTask.class)
+    @JsonSubTypes.Type(name = "reingest", value = ReIngestTask.class)
 })
 public interface Task
 {

indexing-service/src/main/java/io/druid/indexing/firehose/IngestSegmentFirehoseFactory.java

@@ -42,12 +42,9 @@ import io.druid.data.input.InputRow;
 import io.druid.data.input.MapBasedInputRow;
 import io.druid.data.input.impl.InputRowParser;
 import io.druid.granularity.QueryGranularity;
-import io.druid.indexing.common.TaskStatus;
 import io.druid.indexing.common.TaskToolbox;
 import io.druid.indexing.common.TaskToolboxFactory;
 import io.druid.indexing.common.actions.SegmentListUsedAction;
-import io.druid.indexing.common.actions.TaskActionClient;
-import io.druid.indexing.common.task.AbstractTask;
 import io.druid.query.filter.DimFilter;
 import io.druid.query.select.EventHolder;
 import io.druid.segment.Cursor;
@@ -139,10 +136,10 @@ public class IngestSegmentFirehoseFactory implements FirehoseFactory<InputRowPar
   @Override
   public Firehose connect(InputRowParser inputRowParser) throws IOException, ParseException
   {
-    log.info("Connecting firehose: IngestSegmentFirehose[%s,%s]", dataSource, interval);
+    log.info("Connecting firehose: dataSource[%s], interval[%s]", dataSource, interval);
     // better way to achieve this is to pass toolbox to Firehose, The instance is initialized Lazily on connect method.
     final TaskToolbox toolbox = injector.getInstance(TaskToolboxFactory.class).build(
-        new IngestTask("Ingest-Task-Id", dataSource)
+        new ReIngestTask("Ingest-Task-Id", dataSource)
     );
 
     try {

indexing-service/src/main/java/io/druid/indexing/firehose/ReIngestTask.java

@@ -25,9 +25,9 @@ import io.druid.indexing.common.TaskToolbox;
 import io.druid.indexing.common.actions.TaskActionClient;
 import io.druid.indexing.common.task.AbstractTask;
 
-public class IngestTask extends AbstractTask
+public class ReIngestTask extends AbstractTask
 {
-  public IngestTask(
+  public ReIngestTask(
       @JsonProperty("id") final String id,
       @JsonProperty("dataSource") final String dataSource
   )
@@ -38,7 +38,7 @@ public class IngestTask extends AbstractTask
   @Override
   public String getType()
   {
-    return "Ingest-Task";
+    return "reingest";
   }
 
   @Override

server/src/main/java/io/druid/segment/realtime/firehose/CombiningFirehoseFactory.java

@@ -23,6 +23,7 @@ import com.fasterxml.jackson.annotation.JsonCreator;
 import com.fasterxml.jackson.annotation.JsonProperty;
 import com.google.api.client.repackaged.com.google.common.base.Preconditions;
 import com.google.api.client.repackaged.com.google.common.base.Throwables;
+import com.metamx.emitter.EmittingLogger;
 import io.druid.data.input.Firehose;
 import io.druid.data.input.FirehoseFactory;
 import io.druid.data.input.InputRow;
@@ -37,6 +38,8 @@ import java.util.List;
  */
 public class CombiningFirehoseFactory implements FirehoseFactory<InputRowParser>
 {
+  private static final EmittingLogger log = new EmittingLogger(CombiningFirehoseFactory.class);
+
   private final List<FirehoseFactory> delegateFactoryList;
 
   @JsonCreator
@@ -86,10 +89,20 @@ public class CombiningFirehoseFactory implements FirehoseFactory<InputRowParser>
           if (currentFirehose != null) {
             currentFirehose.close();
           }
+
           currentFirehose = firehoseFactoryIterator.next().connect(parser);
         }
         catch (IOException e) {
-          Throwables.propagate(e);
+          if (currentFirehose != null) {
+            try {
+              currentFirehose.close();
+            }
+            catch (IOException e2) {
+              log.error(e, "Unable to close currentFirehose!");
+              throw Throwables.propagate(e2);
+            }
+          }
+          throw Throwables.propagate(e);
         }
       }
     }

server/src/test/java/io/druid/realtime/firehose/CombiningFirehoseFactoryTest.java

@@ -42,16 +42,15 @@ public class CombiningFirehoseFactoryTest
   public void testCombiningfirehose() throws IOException
   {
     List<InputRow> list1 = Arrays.asList(makeRow(1, 1), makeRow(2, 2));
-    List<InputRow> list2 = Arrays.asList(makeRow(3, 3), makeRow(4, 4));
+    List<InputRow> list2 = Arrays.asList(makeRow(3, 3), makeRow(4, 4), makeRow(5, 5));
     FirehoseFactory combiningFactory = new CombiningFirehoseFactory(
         Arrays.<FirehoseFactory>asList(
-            new ListFirehoseFactory(
-                list1
-            ), new ListFirehoseFactory(list2)
+            new ListFirehoseFactory(list1),
+            new ListFirehoseFactory(list2)
         )
     );
     final Firehose firehose = combiningFactory.connect(null);
-    for (int i = 1; i < 5; i++) {
+    for (int i = 1; i < 6; i++) {
       Assert.assertTrue(firehose.hasMore());
       final InputRow inputRow = firehose.nextRow();
       Assert.assertEquals(i, inputRow.getTimestampFromEpoch());
@@ -133,7 +132,7 @@ public class CombiningFirehoseFactoryTest
         @Override
         public void close() throws IOException
         {
-          //
+          // Do nothing
         }
       };
     }