More controllable realtime shutdown

- Realtime plumber will start persisting segments shortly after finishJob is called, regardless
  of rejection policy

- Add "none" rejection policy

realtime/src/main/java/com/metamx/druid/realtime/plumber/NoopRejectionPolicyFactory.java

+package com.metamx.druid.realtime.plumber;
+
+import org.joda.time.DateTime;
+import org.joda.time.Period;
+
+public class NoopRejectionPolicyFactory implements RejectionPolicyFactory
+{
+  @Override
+  public RejectionPolicy create(Period windowPeriod)
+  {
+    return new RejectionPolicy()
+    {
+      @Override
+      public DateTime getCurrMaxTime()
+      {
+        return new DateTime(0);
+      }
+
+      @Override
+      public boolean accept(long timestamp)
+      {
+        return true;
+      }
+    };
+  }
+}

realtime/src/main/java/com/metamx/druid/realtime/plumber/RealtimePlumberSchool.java

@@ -105,6 +105,8 @@ public class RealtimePlumberSchool implements PlumberSchool
   private volatile SegmentPublisher segmentPublisher = null;
   private volatile ServerView serverView = null;
 
+  private volatile boolean noMoreData = false;
+
   @JsonCreator
   public RealtimePlumberSchool(
       @JsonProperty("windowPeriod") Period windowPeriod,
@@ -324,6 +326,8 @@ public class RealtimePlumberSchool implements PlumberSchool
       {
         log.info("Shutting down...");
 
+        noMoreData = true;
+
         while (!sinks.isEmpty()) {
           try {
             log.info(
@@ -553,7 +557,7 @@ public class RealtimePlumberSchool implements PlumberSchool
                     List<Map.Entry<Long, Sink>> sinksToPush = Lists.newArrayList();
                     for (Map.Entry<Long, Sink> entry : sinks.entrySet()) {
                       final Long intervalStart = entry.getKey();
-                      if (intervalStart < minTimestamp) {
+                      if (noMoreData || intervalStart < minTimestamp) {
                         log.info("Adding entry[%s] for merge and push.", entry);
                         sinksToPush.add(entry);
                       }

realtime/src/main/java/com/metamx/druid/realtime/plumber/RejectionPolicyFactory.java

@@ -7,7 +7,8 @@ import org.joda.time.Period;
 @JsonTypeInfo(use = JsonTypeInfo.Id.NAME, property = "type")
 @JsonSubTypes(value = {
     @JsonSubTypes.Type(name = "serverTime", value = ServerTimeRejectionPolicyFactory.class),
-    @JsonSubTypes.Type(name = "messageTime", value = MessageTimeRejectionPolicyFactory.class)
+    @JsonSubTypes.Type(name = "messageTime", value = MessageTimeRejectionPolicyFactory.class),
+    @JsonSubTypes.Type(name = "none", value = NoopRejectionPolicyFactory.class)
 })
 public interface RejectionPolicyFactory
 {