add interface to auto clean mysql pendingSegments table

api/src/main/java/io/druid/timeline/TaskDataSegment.java

@@ -0,0 +1,94 @@
+package io.druid.timeline;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import org.joda.time.Interval;
+
+import java.util.Map;
+
+/**
+ * Created by haoxiang on 16/11/11.
+ */
+public class TaskDataSegment
+{
+
+    private final String type;
+    private final String id;
+    private final String groupId;
+    private final String dataSource;
+    private final Map<String, Object> ioConfig;
+    private final Interval interval;
+
+    @JsonCreator
+    public TaskDataSegment(
+            @JsonProperty("type") String type,
+            @JsonProperty("id") String id,
+            @JsonProperty("groupId") String groupId,
+            @JsonProperty("dataSource") String dataSource,
+            @JsonProperty("ioConfig") Map<String, Object> ioConfig,
+            @JsonProperty("interval") Interval interval
+    )
+    {
+        this.type = type;
+        this.id = id;
+        this.groupId = groupId;
+        this.dataSource = dataSource;
+        this.ioConfig = ioConfig;
+        this.interval = interval;
+    }
+
+    /**
+     * Get dataSource
+     *
+     * @return the dataSource
+     */
+
+    @JsonProperty
+    public String getType()
+    {
+        return type;
+    }
+
+    @JsonProperty
+    public String getId()
+    {
+        return id;
+    }
+
+    @JsonProperty
+    public String getGroupId()
+    {
+        return groupId;
+    }
+
+    @JsonProperty
+    public String getDataSource()
+    {
+        return dataSource;
+    }
+
+    @JsonProperty
+    public Map<String, Object> getIoConfig()
+    {
+        return ioConfig;
+    }
+
+    @JsonProperty
+    public Interval getInterval()
+    {
+        return interval;
+    }
+
+    @Override
+    public String toString()
+    {
+        return "TaskDataSegment{" +
+                "type=" + type +
+                ", id=" + id +
+                ", groupId=" + groupId +
+                ", dataSource=" + dataSource +
+                '}';
+    }
+
+
+}

indexing-service/src/test/java/io/druid/indexing/test/TestIndexerMetadataStorageCoordinator.java

@@ -28,6 +28,7 @@
 import io.druid.indexing.overlord.SegmentPublishResult;
 import io.druid.segment.realtime.appenderator.SegmentIdentifier;
 import io.druid.timeline.DataSegment;
+import io.druid.timeline.TaskDataSegment;
 import org.joda.time.Interval;
 
 import java.io.IOException;
@@ -39,10 +40,12 @@ public class TestIndexerMetadataStorageCoordinator implements IndexerMetadataSto
   final private Set<DataSegment> published = Sets.newConcurrentHashSet();
   final private Set<DataSegment> nuked = Sets.newConcurrentHashSet();
   final private List<DataSegment> unusedSegments;
+  final private List<TaskDataSegment> taskDataSegments;
 
   public TestIndexerMetadataStorageCoordinator()
   {
     unusedSegments = Lists.newArrayList();
+    taskDataSegments = Lists.newArrayList();
   }
 
   @Override
@@ -65,7 +68,7 @@ public List<DataSegment> getUsedSegmentsForInterval(String dataSource, Interval
 
   @Override
   public List<DataSegment> getUsedSegmentsForIntervals(
-      String dataSource, List<Interval> intervals
+          String dataSource, List<Interval> intervals
   ) throws IOException
   {
     return ImmutableList.of();
@@ -93,9 +96,9 @@ public Set<DataSegment> announceHistoricalSegments(Set<DataSegment> segments)
 
   @Override
   public SegmentPublishResult announceHistoricalSegments(
-      Set<DataSegment> segments,
-      DataSourceMetadata oldCommitMetadata,
-      DataSourceMetadata newCommitMetadata
+          Set<DataSegment> segments,
+          DataSourceMetadata oldCommitMetadata,
+          DataSourceMetadata newCommitMetadata
   ) throws IOException
   {
     // Don't actually compare metadata, just do it!
@@ -104,11 +107,11 @@ public SegmentPublishResult announceHistoricalSegments(
 
   @Override
   public SegmentIdentifier allocatePendingSegment(
-      String dataSource,
-      String sequenceName,
-      String previousSegmentId,
-      Interval interval,
-      String maxVersion
+          String dataSource,
+          String sequenceName,
+          String previousSegmentId,
+          Interval interval,
+          String maxVersion
   ) throws IOException
   {
     throw new UnsupportedOperationException();
@@ -120,6 +123,12 @@ public void deleteSegments(Set<DataSegment> segments)
     nuked.addAll(segments);
   }
 
+  @Override
+  public List<TaskDataSegment> getNotActiveTask(final Interval interval){
+
+    return ImmutableList.copyOf(taskDataSegments);
+  }
+
   @Override
   public void updateSegmentMetadata(Set<DataSegment> segments) throws IOException
   {

server/src/main/java/io/druid/indexing/overlord/IndexerMetadataStorageCoordinator.java

@@ -21,6 +21,7 @@
 
 import io.druid.segment.realtime.appenderator.SegmentIdentifier;
 import io.druid.timeline.DataSegment;
+import io.druid.timeline.TaskDataSegment;
 import org.joda.time.Interval;
 
 import java.io.IOException;
@@ -42,7 +43,7 @@ public interface IndexerMetadataStorageCoordinator
    * @throws IOException
    */
   List<DataSegment> getUsedSegmentsForInterval(String dataSource, Interval interval)
-      throws IOException;
+          throws IOException;
 
   /**
    * Get all segments which may include any data in the interval and are flagged as used.
@@ -55,7 +56,7 @@ List<DataSegment> getUsedSegmentsForInterval(String dataSource, Interval interva
    * @throws IOException
    */
   List<DataSegment> getUsedSegmentsForIntervals(final String dataSource, final List<Interval> intervals)
-      throws IOException;
+          throws IOException;
 
   /**
    * Attempts to insert a set of segments to the metadata storage. Returns the set of segments actually added (segments
@@ -86,11 +87,11 @@ List<DataSegment> getUsedSegmentsForIntervals(final String dataSource, final Lis
    * @return the pending segment identifier, or null if it was impossible to allocate a new segment
    */
   SegmentIdentifier allocatePendingSegment(
-      String dataSource,
-      String sequenceName,
-      String previousSegmentId,
-      Interval interval,
-      String maxVersion
+          String dataSource,
+          String sequenceName,
+          String previousSegmentId,
+          Interval interval,
+          String maxVersion
   ) throws IOException;
 
   /**
@@ -113,9 +114,9 @@ SegmentIdentifier allocatePendingSegment(
    * @throws IllegalArgumentException if startMetadata and endMetadata are not either both null or both non-null
    */
   SegmentPublishResult announceHistoricalSegments(
-      Set<DataSegment> segments,
-      DataSourceMetadata startMetadata,
-      DataSourceMetadata endMetadata
+          Set<DataSegment> segments,
+          DataSourceMetadata startMetadata,
+          DataSourceMetadata endMetadata
   ) throws IOException;
 
   DataSourceMetadata getDataSourceMetadata(String dataSource);
@@ -141,4 +142,11 @@ SegmentPublishResult announceHistoricalSegments(
    * @return DataSegments which include ONLY data within the requested interval and are not flagged as used. Data segments NOT returned here may include data in the interval
    */
   List<DataSegment> getUnusedSegmentsForInterval(String dataSource, Interval interval);
+
+  /**
+   * Get all task which active property is 0, is used to delete the useless segments in pendingSegment table.
+   * @param interval Filter the tasks to ones that include tasks in this interval exclusively. Start is inclusive, end is exclusive
+   * @return The TaskDataSegment list which used to match the pendingSegment table's payload
+   */
+  List<TaskDataSegment> getNotActiveTask(final Interval interval);
 }