ElasticSearchRestDAOV5 implements IndexDAO { @Override public List<String> searchArchivableWorkflows(String indexName, long archiveTtlDays) { QueryBuilder q = QueryBuilders.boolQuery() .must(QueryBuilders.rangeQuery("endTime").lt(LocalDate.now(ZoneOffset.UTC).minusDays(archiveTtlDays).toString()).gte(LocalDate.now(ZoneOffset.UTC).minusDays(archiveTtlDays).minusDays(1).toString())) .should(QueryBuilders.termQuery("status", "COMPLETED")) .should(QueryBuilders.termQuery("status", "FAILED")) .should(QueryBuilders.termQuery("status", "TIMED_OUT")) .should(QueryBuilders.termQuery("status", "TERMINATED")) .mustNot(QueryBuilders.existsQuery("archived")) .minimumShouldMatch(1); SearchResult<String> workflowIds; try { workflowIds = searchObjectIds(indexName, q, 0, 1000, null, WORKFLOW_DOC_TYPE); } catch (IOException e) { logger.error("Unable to communicate with ES to find archivable 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 testSearchArchivableWorkflows() throws IOException { String workflowId = "search-workflow-id"; Long time = DateTime.now(DateTimeZone.UTC).minusDays(7).toDate().getTime(); workflow.setWorkflowId(workflowId); workflow.setStatus(Workflow.WorkflowStatus.COMPLETED); workflow.setCreateTime(time); workflow.setUpdateTime(time); workflow.setEndTime(time); indexDAO.indexWorkflow(workflow); assertTrue(indexExists("conductor")); await() .atMost(5, TimeUnit.SECONDS) .untilAsserted( () -> { List<String> searchIds = indexDAO.searchArchivableWorkflows("conductor",6); assertEquals(1, searchIds.size()); assertEquals(workflowId, searchIds.get(0)); } ); }