ElasticSearchRestDAOV6 extends ElasticSearchBaseDAO implements IndexDAO { @Override public void indexWorkflow(Workflow workflow) { try { long startTime = Instant.now().toEpochMilli(); String workflowId = workflow.getWorkflowId(); WorkflowSummary summary = new WorkflowSummary(workflow); byte[] docBytes = objectMapper.writeValueAsBytes(summary); String docType = StringUtils.isBlank(docTypeOverride) ? WORKFLOW_DOC_TYPE : docTypeOverride; IndexRequest request = new IndexRequest(workflowIndexName, docType, workflowId); request.source(docBytes, XContentType.JSON); new RetryUtil<IndexResponse>().retryOnException(() -> { try { return elasticSearchClient.index(request); } catch (IOException e) { throw new RuntimeException(e); } }, null, null, RETRY_COUNT, "Indexing workflow document: " + workflow.getWorkflowId(), "indexWorkflow"); long endTime = Instant.now().toEpochMilli(); logger.debug("Time taken {} for indexing workflow: {}", endTime - startTime, workflowId); Monitors.recordESIndexTime("index_workflow", WORKFLOW_DOC_TYPE, endTime - startTime); Monitors.recordWorkerQueueSize("indexQueue", ((ThreadPoolExecutor) executorService).getQueue().size()); } catch (Exception e) { Monitors.error(className, "indexWorkflow"); logger.error("Failed to index workflow: {}", workflow.getWorkflowId(), e); } } @Inject  ElasticSearchRestDAOV6(RestClientBuilder restClientBuilder, 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 List<Message> getMessages(String queue); @Override List<EventExecution> getEventExecutions(String event); @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);  }
@Test public void shouldIndexWorkflow() { Workflow workflow = TestUtils.loadWorkflowSnapshot("workflow"); WorkflowSummary summary = new WorkflowSummary(workflow); indexDAO.indexWorkflow(workflow); assertWorkflowSummary(workflow.getWorkflowId(), summary); }