Apache
/
druid
mirror of https://github.com/apache/druid.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Added support for filtering by unused parameter for HeapMemoryTaskStorage (#6510) 

* 1. added support for unused DateTime start parameter in getRecentlyFinishedTaskInfoSince method:
 HeapMemoryTaskStorage.getRecentlyFinishedTaskInfoSince return the finished tasks by comparing TaskStuff.createdDate with the start time
2. added filtering by status complete to TaskStuff list stream in HeapMemoryTaskStorage.getNRecentlyFinishedTaskInfo method.
3. changed names of methods and parameters to present that public API method OverlordResource.getTasks return the list of completed tasks, which createdDate, not date of completion, belongs to the interval parameter.

* 1. added support for unused DateTime start parameter in getRecentlyFinishedTaskInfoSince method:
 HeapMemoryTaskStorage.getRecentlyFinishedTaskInfoSince return the finished tasks by comparing TaskStuff.createdDate with the start time
2. added filtering by status complete to TaskStuff list stream in HeapMemoryTaskStorage.getNRecentlyFinishedTaskInfo method.
3. changed names of methods and parameters to present that public API method OverlordResource.getTasks return the list of completed tasks, which createdDate, not date of completion, belongs to the interval parameter.

* Fixed OverlordResourceTest to Support changed methods names

* Changed methods and parameters names to make them more obvious to understand.

* Changed String.replace() for the StringUtils.replace()(#6607)

* Fixed checkstyle error
This commit is contained in:
Marat 2018-11-21 00:42:44 +03:00 committed by Jihoon Son
parent 87b96fb1fd
commit 81c9a6177c
6 changed files with 30 additions and 29 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

17
indexing-service/src/main/java/org/apache/druid/indexing/overlord/HeapMemoryTaskStorage.java
View File

@ -221,9 +221,9 @@ public class HeapMemoryTaskStorage implements TaskStorage
} }
@Override @Override
public List<TaskInfo<Task, TaskStatus>> getRecentlyFinishedTaskInfo( public List<TaskInfo<Task, TaskStatus>> getRecentlyCreatedAlreadyFinishedTaskInfo(
@Nullable Integer maxTaskStatuses, @Nullable Integer maxTaskStatuses,
@Nullable Duration duration, @Nullable Duration durationBeforeNow,
@Nullable String datasource @Nullable String datasource
) )
{ {
@ -240,18 +240,18 @@ public class HeapMemoryTaskStorage implements TaskStorage
}.reverse(); }.reverse();
return maxTaskStatuses == null ? return maxTaskStatuses == null ?
getRecentlyFinishedTaskInfoSince( getRecentlyCreatedAlreadyFinishedTaskInfoSince(
DateTimes.nowUtc().minus(duration == null ? config.getRecentlyFinishedThreshold() : duration), DateTimes.nowUtc().minus(durationBeforeNow == null ? config.getRecentlyFinishedThreshold() : durationBeforeNow),
createdDateDesc createdDateDesc
) : ) :
getNRecentlyFinishedTaskInfo(maxTaskStatuses, createdDateDesc); getNRecentlyCreatedAlreadyFinishedTaskInfo(maxTaskStatuses, createdDateDesc);
} }
finally { finally {
giant.unlock(); giant.unlock();
} }
} }
private List<TaskInfo<Task, TaskStatus>> getRecentlyFinishedTaskInfoSince( private List<TaskInfo<Task, TaskStatus>> getRecentlyCreatedAlreadyFinishedTaskInfoSince(
DateTime start, DateTime start,
Ordering<TaskStuff> createdDateDesc Ordering<TaskStuff> createdDateDesc
) )
@ -262,7 +262,7 @@ public class HeapMemoryTaskStorage implements TaskStorage
List<TaskStuff> list = createdDateDesc List<TaskStuff> list = createdDateDesc
.sortedCopy(tasks.values()) .sortedCopy(tasks.values())
.stream() .stream()
.filter(taskStuff -> taskStuff.getStatus().isComplete()) .filter(taskStuff -> taskStuff.getStatus().isComplete() && taskStuff.createdDate.isAfter(start))
.collect(Collectors.toList()); .collect(Collectors.toList());
final ImmutableList.Builder<TaskInfo<Task, TaskStatus>> listBuilder = ImmutableList.builder(); final ImmutableList.Builder<TaskInfo<Task, TaskStatus>> listBuilder = ImmutableList.builder();
for (final TaskStuff taskStuff : list) { for (final TaskStuff taskStuff : list) {
@ -283,7 +283,7 @@ public class HeapMemoryTaskStorage implements TaskStorage
} }
} }
private List<TaskInfo<Task, TaskStatus>> getNRecentlyFinishedTaskInfo(int n, Ordering<TaskStuff> createdDateDesc) private List<TaskInfo<Task, TaskStatus>> getNRecentlyCreatedAlreadyFinishedTaskInfo(int n, Ordering<TaskStuff> createdDateDesc)
{ {
giant.lock(); giant.lock();
@ -291,6 +291,7 @@ public class HeapMemoryTaskStorage implements TaskStorage
List<TaskStuff> list = createdDateDesc List<TaskStuff> list = createdDateDesc
.sortedCopy(tasks.values()) .sortedCopy(tasks.values())
.stream() .stream()
.filter(taskStuff -> taskStuff.getStatus().isComplete())
.limit(n) .limit(n)
.collect(Collectors.toList()); .collect(Collectors.toList());
final ImmutableList.Builder<TaskInfo<Task, TaskStatus>> listBuilder = ImmutableList.builder(); final ImmutableList.Builder<TaskInfo<Task, TaskStatus>> listBuilder = ImmutableList.builder();

6
indexing-service/src/main/java/org/apache/druid/indexing/overlord/MetadataTaskStorage.java
View File

@ -208,15 +208,15 @@ public class MetadataTaskStorage implements TaskStorage
} }
@Override @Override
public List<TaskInfo<Task, TaskStatus>> getRecentlyFinishedTaskInfo( public List<TaskInfo<Task, TaskStatus>> getRecentlyCreatedAlreadyFinishedTaskInfo(
@Nullable Integer maxTaskStatuses, @Nullable Integer maxTaskStatuses,
@Nullable Duration duration, @Nullable Duration durationBeforeNow,
@Nullable String datasource @Nullable String datasource
) )
{ {
return ImmutableList.copyOf( return ImmutableList.copyOf(
handler.getCompletedTaskInfo( handler.getCompletedTaskInfo(
DateTimes.nowUtc().minus(duration == null ? config.getRecentlyFinishedThreshold() : duration), DateTimes.nowUtc().minus(durationBeforeNow == null ? config.getRecentlyFinishedThreshold() : durationBeforeNow),
maxTaskStatuses, maxTaskStatuses,
datasource datasource
) )

6
indexing-service/src/main/java/org/apache/druid/indexing/overlord/TaskStorage.java
View File

@ -141,14 +141,14 @@ public interface TaskStorage
* return nothing. * return nothing.
* *
* @param maxTaskStatuses maxTaskStatuses * @param maxTaskStatuses maxTaskStatuses
* @param duration duration * @param durationBeforeNow duration
* @param datasource datasource * @param datasource datasource
* *
* @return list of {@link TaskInfo} * @return list of {@link TaskInfo}
*/ */
List<TaskInfo<Task, TaskStatus>> getRecentlyFinishedTaskInfo( List<TaskInfo<Task, TaskStatus>> getRecentlyCreatedAlreadyFinishedTaskInfo(
@Nullable Integer maxTaskStatuses, @Nullable Integer maxTaskStatuses,
@Nullable Duration duration, @Nullable Duration durationBeforeNow,
@Nullable String datasource @Nullable String datasource
); );

4
indexing-service/src/main/java/org/apache/druid/indexing/overlord/TaskStorageQueryAdapter.java
View File

@ -58,13 +58,13 @@ public class TaskStorageQueryAdapter
return storage.getActiveTaskInfo(dataSource); return storage.getActiveTaskInfo(dataSource);
} }
public List<TaskInfo<Task, TaskStatus>> getRecentlyCompletedTaskInfo( public List<TaskInfo<Task, TaskStatus>> getCompletedTaskInfoByCreatedTimeDuration(
@Nullable Integer maxTaskStatuses, @Nullable Integer maxTaskStatuses,
@Nullable Duration duration, @Nullable Duration duration,
@Nullable String dataSource @Nullable String dataSource
) )
{ {
return storage.getRecentlyFinishedTaskInfo(maxTaskStatuses, duration, dataSource); return storage.getRecentlyCreatedAlreadyFinishedTaskInfo(maxTaskStatuses, duration, dataSource);
} }
public Optional<Task> getTask(final String taskid) public Optional<Task> getTask(final String taskid)

12
indexing-service/src/main/java/org/apache/druid/indexing/overlord/http/OverlordResource.java
View File

@ -627,7 +627,7 @@ public class OverlordResource
public Response getTasks( public Response getTasks(
@QueryParam("state") final String state, @QueryParam("state") final String state,
@QueryParam("datasource") final String dataSource, @QueryParam("datasource") final String dataSource,
@QueryParam("interval") final String interval, @PathParam("createdTimeInterval") final String createdTimeInterval,
@QueryParam("max") final Integer maxCompletedTasks, @QueryParam("max") final Integer maxCompletedTasks,
@QueryParam("type") final String type, @QueryParam("type") final String type,
@Context final HttpServletRequest req @Context final HttpServletRequest req
@ -692,13 +692,13 @@ public class OverlordResource
//checking for complete tasks first to avoid querying active tasks if user only wants complete tasks //checking for complete tasks first to avoid querying active tasks if user only wants complete tasks
if (state == null || "complete".equals(StringUtils.toLowerCase(state))) { if (state == null || "complete".equals(StringUtils.toLowerCase(state))) {
Duration duration = null; Duration createdTimeDuration = null;
if (interval != null) { if (createdTimeInterval != null) {
final Interval theInterval = Intervals.of(interval.replace('_', '/')); final Interval theInterval = Intervals.of(StringUtils.replace(createdTimeInterval, "_", "/"));
duration = theInterval.toDuration(); createdTimeDuration = theInterval.toDuration();
} }
final List<TaskInfo<Task, TaskStatus>> taskInfoList = final List<TaskInfo<Task, TaskStatus>> taskInfoList =
taskStorageQueryAdapter.getRecentlyCompletedTaskInfo(maxCompletedTasks, duration, dataSource); taskStorageQueryAdapter.getCompletedTaskInfoByCreatedTimeDuration(maxCompletedTasks, createdTimeDuration, dataSource);
final List<TaskStatusPlus> completedTasks = taskInfoList.stream() final List<TaskStatusPlus> completedTasks = taskInfoList.stream()
.map(completeTaskTransformFunc::apply) .map(completeTaskTransformFunc::apply)
.collect(Collectors.toList()); .collect(Collectors.toList());

14
indexing-service/src/test/java/org/apache/druid/indexing/overlord/http/OverlordResourceTest.java
View File

@ -233,7 +233,7 @@ public class OverlordResourceTest
new MockTaskRunnerWorkItem(tasksIds.get(1), null), new MockTaskRunnerWorkItem(tasksIds.get(1), null),
new MockTaskRunnerWorkItem(tasksIds.get(2), null))); new MockTaskRunnerWorkItem(tasksIds.get(2), null)));
EasyMock.expect(taskStorageQueryAdapter.getRecentlyCompletedTaskInfo(null, null, null)).andStubReturn( EasyMock.expect(taskStorageQueryAdapter.getCompletedTaskInfoByCreatedTimeDuration(null, null, null)).andStubReturn(
ImmutableList.of( ImmutableList.of(
new TaskInfo( new TaskInfo(
"id_1", "id_1",
@ -259,7 +259,7 @@ public class OverlordResourceTest
) )
); );
EasyMock.replay(taskRunner, taskMaster, taskStorageQueryAdapter, indexerMetadataStorageAdapter, req); EasyMock.replay(taskRunner, taskMaster, taskStorageQueryAdapter, indexerMetadataStorageAdapter, req);
Assert.assertTrue(taskStorageQueryAdapter.getRecentlyCompletedTaskInfo(null, null, null).size() == 3); Assert.assertTrue(taskStorageQueryAdapter.getCompletedTaskInfoByCreatedTimeDuration(null, null, null).size() == 3);
Assert.assertTrue(taskRunner.getRunningTasks().size() == 3); Assert.assertTrue(taskRunner.getRunningTasks().size() == 3);
List<TaskStatusPlus> responseObjects = (List) overlordResource List<TaskStatusPlus> responseObjects = (List) overlordResource
.getCompleteTasks(null, req).getEntity(); .getCompleteTasks(null, req).getEntity();
@ -313,7 +313,7 @@ public class OverlordResourceTest
{ {
expectAuthorizationTokenCheck(); expectAuthorizationTokenCheck();
//completed tasks //completed tasks
EasyMock.expect(taskStorageQueryAdapter.getRecentlyCompletedTaskInfo(null, null, null)).andStubReturn( EasyMock.expect(taskStorageQueryAdapter.getCompletedTaskInfoByCreatedTimeDuration(null, null, null)).andStubReturn(
ImmutableList.of( ImmutableList.of(
new TaskInfo( new TaskInfo(
"id_5", "id_5",
@ -403,7 +403,7 @@ public class OverlordResourceTest
{ {
expectAuthorizationTokenCheck(); expectAuthorizationTokenCheck();
//completed tasks //completed tasks
EasyMock.expect(taskStorageQueryAdapter.getRecentlyCompletedTaskInfo(null, null, "allow")).andStubReturn( EasyMock.expect(taskStorageQueryAdapter.getCompletedTaskInfoByCreatedTimeDuration(null, null, "allow")).andStubReturn(
ImmutableList.of( ImmutableList.of(
new TaskInfo( new TaskInfo(
"id_5", "id_5",
@ -667,7 +667,7 @@ public class OverlordResourceTest
public void testGetTasksFilterCompleteState() public void testGetTasksFilterCompleteState()
{ {
expectAuthorizationTokenCheck(); expectAuthorizationTokenCheck();
EasyMock.expect(taskStorageQueryAdapter.getRecentlyCompletedTaskInfo(null, null, null)).andStubReturn( EasyMock.expect(taskStorageQueryAdapter.getCompletedTaskInfoByCreatedTimeDuration(null, null, null)).andStubReturn(
ImmutableList.of( ImmutableList.of(
new TaskInfo( new TaskInfo(
"id_1", "id_1",
@ -707,7 +707,7 @@ public class OverlordResourceTest
expectAuthorizationTokenCheck(); expectAuthorizationTokenCheck();
List<String> tasksIds = ImmutableList.of("id_1", "id_2", "id_3"); List<String> tasksIds = ImmutableList.of("id_1", "id_2", "id_3");
Duration duration = new Period("PT86400S").toStandardDuration(); Duration duration = new Period("PT86400S").toStandardDuration();
EasyMock.expect(taskStorageQueryAdapter.getRecentlyCompletedTaskInfo(null, duration, null)).andStubReturn( EasyMock.expect(taskStorageQueryAdapter.getCompletedTaskInfoByCreatedTimeDuration(null, duration, null)).andStubReturn(
ImmutableList.of( ImmutableList.of(
new TaskInfo( new TaskInfo(
"id_1", "id_1",
@ -747,7 +747,7 @@ public class OverlordResourceTest
public void testGetNullCompleteTask() public void testGetNullCompleteTask()
{ {
expectAuthorizationTokenCheck(); expectAuthorizationTokenCheck();
EasyMock.expect(taskStorageQueryAdapter.getRecentlyCompletedTaskInfo(null, null, null)).andStubReturn( EasyMock.expect(taskStorageQueryAdapter.getCompletedTaskInfoByCreatedTimeDuration(null, null, null)).andStubReturn(
ImmutableList.of( ImmutableList.of(
new TaskInfo( new TaskInfo(
"id_1", "id_1",