Merge pull request #641 from metamx/cleanup-ingest

some minor cleanups to ingest firehose

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

@@ -24,7 +24,6 @@ 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.query.Query;
 import io.druid.query.QueryRunner;
 
@@ -54,8 +53,7 @@ import io.druid.query.QueryRunner;
     @JsonSubTypes.Type(name = "index_realtime", value = RealtimeIndexTask.class),
     @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 = "version_converter_sub", value = VersionConverterTask.SubTask.class)
 })
 public interface Task
 {

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

@@ -42,12 +42,10 @@ 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.indexing.common.task.NoopTask;
 import io.druid.query.filter.DimFilter;
 import io.druid.query.select.EventHolder;
 import io.druid.segment.Cursor;
@@ -139,10 +137,11 @@ 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.
+    // Noop Task is just used to create the toolbox and list segments.
     final TaskToolbox toolbox = injector.getInstance(TaskToolboxFactory.class).build(
-        new IngestTask("Ingest-Task-Id", dataSource)
+        new NoopTask("reingest", 0, 0, null, null)
     );
 
     try {

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

-/*
- * Druid - a distributed column store.
- * Copyright (C) 2012, 2013, 2014  Metamarkets Group Inc.
- *
- * This program is free software; you can redistribute it and/or
- * modify it under the terms of the GNU General Public License
- * as published by the Free Software Foundation; either version 2
- * of the License, or (at your option) any later version.
- *
- * This program is distributed in the hope that it will be useful,
- * but WITHOUT ANY WARRANTY; without even the implied warranty of
- * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
- * GNU General Public License for more details.
- *
- * You should have received a copy of the GNU General Public License
- * along with this program; if not, write to the Free Software
- * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA.
- */
-
-package io.druid.indexing.firehose;
-
-import com.fasterxml.jackson.annotation.JsonProperty;
-import io.druid.indexing.common.TaskStatus;
-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 IngestTask(
-      @JsonProperty("id") final String id,
-      @JsonProperty("dataSource") final String dataSource
-  )
-  {
-    super(id, dataSource);
-  }
-
-  @Override
-  public String getType()
-  {
-    return "Ingest-Task";
-  }
-
-  @Override
-  public boolean isReady(TaskActionClient taskActionClient) throws Exception
-  {
-    return true;
-  }
-
-  @Override
-  public TaskStatus run(TaskToolbox toolbox) throws Exception
-  {
-    return TaskStatus.success(getId());
-  }
-}
\ No newline at end of file

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

@@ -44,16 +44,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());
@@ -84,12 +83,6 @@ public class CombiningFirehoseFactoryTest
         return new DateTime(timestamp);
       }
 
-      @Override
-      public int compareTo(Row o)
-      {
-        return 0;
-      }
-
       @Override
       public List<String> getDimension(String dimension)
       {
@@ -108,6 +101,11 @@ public class CombiningFirehoseFactoryTest
         return null;
       }
 
+      @Override
+      public int compareTo(Row o)
+      {
+        return 0;
+      }
     };
   }
 
@@ -147,7 +145,7 @@ public class CombiningFirehoseFactoryTest
         @Override
         public void close() throws IOException
         {
-          //
+          // Do nothing
         }
       };
     }