ElasticSearchRestDAOV5 implements IndexDAO { public List<String> searchRecentRunningWorkflows(int lastModifiedHoursAgoFrom, int lastModifiedHoursAgoTo) { DateTime dateTime = new DateTime(); QueryBuilder q = QueryBuilders.boolQuery() .must(QueryBuilders.rangeQuery("updateTime") .gt(dateTime.minusHours(lastModifiedHoursAgoFrom))) .must(QueryBuilders.rangeQuery("updateTime") .lt(dateTime.minusHours(lastModifiedHoursAgoTo))) .must(QueryBuilders.termQuery("status", "RUNNING")); SearchResult<String> workflowIds; try { workflowIds = searchObjectIds(indexName, q, 0, 5000, Collections.singletonList("updateTime:ASC"), WORKFLOW_DOC_TYPE); } catch (IOException e) { logger.error("Unable to communicate with ES to find recent running workflows", e); return Collections.emptyList(); } return workflowIds.getResults(); } @Inject  ElasticSearchRestDAOV5(RestClient lowLevelRestClient, ElasticSearchConfiguration config, ObjectMapper objectMapper); @Override void setup(); boolean doesResourceExist(final String resourcePath); boolean doesResourceNotExist(final String resourcePath); @Override void indexWorkflow(Workflow workflow); @Override CompletableFuture<Void> asyncIndexWorkflow(Workflow workflow); @Override void indexTask(Task task); @Override CompletableFuture<Void> asyncIndexTask(Task task); @Override void addTaskExecutionLogs(List<TaskExecLog> taskExecLogs); @Override CompletableFuture<Void> asyncAddTaskExecutionLogs(List<TaskExecLog> logs); @Override List<TaskExecLog> getTaskExecutionLogs(String taskId); @Override void addMessage(String queue, Message message); @Override CompletableFuture<Void> asyncAddMessage(String queue, Message message); @Override void addEventExecution(EventExecution eventExecution); @Override CompletableFuture<Void> asyncAddEventExecution(EventExecution eventExecution); @Override SearchResult<String> searchWorkflows(String query, String freeText, int start, int count, List<String> sort); @Override SearchResult<String> searchTasks(String query, String freeText, int start, int count, List<String> sort); @Override void removeWorkflow(String workflowId); @Override CompletableFuture<Void> asyncRemoveWorkflow(String workflowId); @Override void updateWorkflow(String workflowInstanceId, String[] keys, Object[] values); @Override CompletableFuture<Void> asyncUpdateWorkflow(String workflowInstanceId, String[] keys, Object[] values); @Override String get(String workflowInstanceId, String fieldToGet); @Override List<String> searchArchivableWorkflows(String indexName, long archiveTtlDays); List<String> searchRecentRunningWorkflows(int lastModifiedHoursAgoFrom, int lastModifiedHoursAgoTo); @Override List<Message> getMessages(String queue); @Override List<EventExecution> getEventExecutions(String event);  }
@Test public void testSearchRecentRunningWorkflows() { workflow.setWorkflowId("completed-workflow"); workflow.setStatus(Workflow.WorkflowStatus.COMPLETED); indexDAO.indexWorkflow(workflow); String workflowId = "recent-running-workflow-id"; workflow.setWorkflowId(workflowId); workflow.setStatus(Workflow.WorkflowStatus.RUNNING); workflow.setCreateTime(new Date().getTime()); workflow.setUpdateTime(new Date().getTime()); workflow.setEndTime(new Date().getTime()); indexDAO.indexWorkflow(workflow); await() .atMost(3, TimeUnit.SECONDS) .untilAsserted( () -> { List<String> searchIds = indexDAO.searchRecentRunningWorkflows(1,0); assertEquals(1, searchIds.size()); assertEquals(workflowId, searchIds.get(0)); } ); }