ElasticSearchDAOV5 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); SearchRequestBuilder s = elasticSearchClient.prepareSearch(indexName) .setTypes("workflow") .setQuery(q) .addSort("endTime", SortOrder.ASC) .setSize(archiveSearchBatchSize); SearchResponse response = s.execute().actionGet(); SearchHits hits = response.getHits(); logger.info("Archive search totalHits - {}", hits.getTotalHits()); return Arrays.stream(hits.getHits()) .map(SearchHit::getId) .collect(Collectors.toCollection(LinkedList::new)); } @Inject  ElasticSearchDAOV5(Client elasticSearchClient, ElasticSearchConfiguration config,
        ObjectMapper objectMapper); @Override void setup(); @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); @Override List<String> searchRecentRunningWorkflows(int lastModifiedHoursAgoFrom,
        int lastModifiedHoursAgoTo); @Override List<Message> getMessages(String queue); @Override List<EventExecution> getEventExecutions(String event);  }
@Test public void testSearchArchivableWorkflows() { String workflowId = "search-workflow-id"; workflow.setWorkflowId(workflowId); workflow.setStatus(Workflow.WorkflowStatus.COMPLETED); workflow.setCreateTime((new Date(System.currentTimeMillis() - TimeUnit.DAYS.toMillis(4))).getTime()); workflow.setUpdateTime((new Date(System.currentTimeMillis() - TimeUnit.DAYS.toMillis(4))).getTime()); workflow.setEndTime((new Date(System.currentTimeMillis() - TimeUnit.DAYS.toMillis(4))).getTime()); indexDAO.indexWorkflow(workflow); await() .atMost(3, TimeUnit.SECONDS) .untilAsserted( () -> { List<String> searchIds = indexDAO.searchArchivableWorkflows("conductor",3); assertEquals(1, searchIds.size()); assertEquals(workflowId, searchIds.get(0)); } ); }