Automatic pendingSegments cleanup

indexing-service/src/main/java/io/druid/indexing/common/TaskLocation.java → api/src/main/java/io/druid/indexer/TaskLocation.java

@@ -17,7 +17,7 @@
  * under the License.
  */
 
-package io.druid.indexing.common;
+package io.druid.indexer;
 
 import com.fasterxml.jackson.annotation.JsonCreator;
 import com.fasterxml.jackson.annotation.JsonProperty;

api/src/main/java/io/druid/indexer/TaskState.java

@@ -0,0 +1,47 @@
+/*
+ * Licensed to Metamarkets Group Inc. (Metamarkets) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. Metamarkets licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package io.druid.indexer;
+
+public enum TaskState
+{
+  RUNNING,
+  SUCCESS,
+  FAILED;
+
+  public boolean isRunnable()
+  {
+    return this == RUNNING;
+  }
+
+  public boolean isComplete()
+  {
+    return this != RUNNING;
+  }
+
+  public boolean isSuccess()
+  {
+    return this == SUCCESS;
+  }
+
+  public boolean isFailure()
+  {
+    return this == FAILED;
+  }
+}

api/src/main/java/io/druid/indexer/TaskStatusPlus.java

@@ -0,0 +1,94 @@
+/*
+ * Licensed to Metamarkets Group Inc. (Metamarkets) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. Metamarkets licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package io.druid.indexer;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.google.common.base.Preconditions;
+import org.joda.time.DateTime;
+
+import javax.annotation.Nullable;
+
+public class TaskStatusPlus
+{
+  private final String id;
+  private final DateTime createdTime;
+  private final DateTime queueInsertionTime;
+  private final TaskState state;
+  private final Long duration;
+  private final TaskLocation location;
+
+  @JsonCreator
+  public TaskStatusPlus(
+      @JsonProperty("id") String id,
+      @JsonProperty("createdTime") DateTime createdTime,
+      @JsonProperty("queueInsertionTime") DateTime queueInsertionTime,
+      @JsonProperty("state") @Nullable TaskState state,
+      @JsonProperty("duration") @Nullable Long duration,
+      @JsonProperty("location") TaskLocation location
+  )
+  {
+    if (state != null && state.isComplete()) {
+      Preconditions.checkNotNull(duration, "duration");
+    }
+    this.id = Preconditions.checkNotNull(id, "id");
+    this.createdTime = Preconditions.checkNotNull(createdTime, "createdTime");
+    this.queueInsertionTime = Preconditions.checkNotNull(queueInsertionTime, "queueInsertionTime");
+    this.state = state;
+    this.duration = duration;
+    this.location = Preconditions.checkNotNull(location, "location");
+  }
+
+  @JsonProperty
+  public String getId()
+  {
+    return id;
+  }
+
+  @JsonProperty
+  public DateTime getCreatedTime()
+  {
+    return createdTime;
+  }
+
+  @JsonProperty
+  public DateTime getQueueInsertionTime()
+  {
+    return queueInsertionTime;
+  }
+
+  @JsonProperty
+  public TaskState getState()
+  {
+    return state;
+  }
+
+  @JsonProperty
+  public Long getDuration()
+  {
+    return duration;
+  }
+
+  @JsonProperty
+  public TaskLocation getLocation()
+  {
+    return location;
+  }
+}

common/src/main/java/io/druid/metadata/MetadataStorageActionHandler.java

@@ -85,13 +85,27 @@ void insert(
    */
   List<Pair<EntryType, StatusType>> getActiveEntriesWithStatus();
 
+  default List<StatusType> getInactiveStatusesSince(DateTime timestamp)
+  {
+    return getInactiveStatusesSince(timestamp, null);
+  }
+
   /**
-   * Return all statuses for inactive entries created on or later than the given timestamp
+   * Return up to {@code maxNumStatuses} statuses for inactive entries created on or later than the given timestamp
    *
    * @param timestamp timestamp
+   * @param maxNumStatuses maxNumStatuses
    * @return list of statuses
    */
-  List<StatusType> getInactiveStatusesSince(DateTime timestamp);
+  List<StatusType> getInactiveStatusesSince(DateTime timestamp, @Nullable Integer maxNumStatuses);
+
+  /**
+   * Return createdDate and dataSource for the given id
+   *
+   * @return a pair of createdDate and dataSource or null if an entry for the given id is not found
+   */
+  @Nullable
+  Pair<DateTime, String> getCreatedDateAndDataSource(String entryId);
 
   /**
    * Add a lock to the given entry

extensions-contrib/sqlserver-metadata-storage/src/main/java/io/druid/metadata/storage/sqlserver/SQLServerMetadataStorageModule.java

@@ -26,10 +26,12 @@
 import io.druid.guice.PolyBind;
 import io.druid.guice.SQLMetadataStorageDruidModule;
 import io.druid.initialization.DruidModule;
+import io.druid.metadata.MetadataStorageActionHandlerFactory;
 import io.druid.metadata.MetadataStorageConnector;
 import io.druid.metadata.MetadataStorageProvider;
 import io.druid.metadata.NoopMetadataStorageProvider;
 import io.druid.metadata.SQLMetadataConnector;
+import io.druid.metadata.SQLServerMetadataStorageActionHandlerFactory;
 
 import java.util.List;
 
@@ -72,5 +74,10 @@ public void configure(Binder binder)
         .addBinding(TYPE)
         .to(SQLServerConnector.class)
         .in(LazySingleton.class);
+
+    PolyBind.optionBinder(binder, Key.get(MetadataStorageActionHandlerFactory.class))
+            .addBinding(TYPE)
+            .to(SQLServerMetadataStorageActionHandlerFactory.class)
+            .in(LazySingleton.class);
   }
 }

extensions-core/kafka-indexing-service/src/main/java/io/druid/indexing/kafka/KafkaIndexTaskClient.java

@@ -38,7 +38,7 @@
 import io.druid.indexing.common.RetryPolicyConfig;
 import io.druid.indexing.common.RetryPolicyFactory;
 import io.druid.indexing.common.TaskInfoProvider;
-import io.druid.indexing.common.TaskLocation;
+import io.druid.indexer.TaskLocation;
 import io.druid.indexing.common.TaskStatus;
 import io.druid.java.util.common.IAE;
 import io.druid.java.util.common.IOE;

extensions-core/kafka-indexing-service/src/main/java/io/druid/indexing/kafka/supervisor/KafkaSupervisor.java

@@ -47,7 +47,7 @@
 import com.metamx.emitter.service.ServiceEmitter;
 import com.metamx.emitter.service.ServiceMetricEvent;
 import io.druid.indexing.common.TaskInfoProvider;
-import io.druid.indexing.common.TaskLocation;
+import io.druid.indexer.TaskLocation;
 import io.druid.indexing.common.TaskStatus;
 import io.druid.indexing.common.task.Task;
 import io.druid.indexing.common.task.TaskResource;

extensions-core/kafka-indexing-service/src/test/java/io/druid/indexing/kafka/KafkaIndexTaskClientTest.java

@@ -32,7 +32,7 @@
 import com.metamx.http.client.response.FullResponseHandler;
 import com.metamx.http.client.response.FullResponseHolder;
 import io.druid.indexing.common.TaskInfoProvider;
-import io.druid.indexing.common.TaskLocation;
+import io.druid.indexer.TaskLocation;
 import io.druid.indexing.common.TaskStatus;
 import io.druid.jackson.DefaultObjectMapper;
 import io.druid.java.util.common.DateTimes;