ElasticSearchRestDAOV5 implements IndexDAO { @Override public void indexTask(Task task) { try { long startTime = Instant.now().toEpochMilli(); String taskId = task.getTaskId(); TaskSummary summary = new TaskSummary(task); indexObject(indexName, TASK_DOC_TYPE, taskId, summary); long endTime = Instant.now().toEpochMilli(); logger.debug("Time taken {} for indexing task:{} in workflow: {}", endTime - startTime, taskId, task.getWorkflowInstanceId()); Monitors.recordESIndexTime("index_task", TASK_DOC_TYPE, endTime - startTime); Monitors.recordWorkerQueueSize("indexQueue", ((ThreadPoolExecutor) executorService).getQueue().size()); } catch (Exception e) { logger.error("Failed to index task: {}", task.getTaskId(), e); } } @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 indexTask() throws Exception { String correlationId = "some-correlation-id"; Task task = new Task(); task.setTaskId("some-task-id"); task.setWorkflowInstanceId("some-workflow-instance-id"); task.setTaskType("some-task-type"); task.setStatus(Status.FAILED); task.setInputData(new HashMap<String, Object>() {{ put("input_key", "input_value"); }}); task.setCorrelationId(correlationId); task.setTaskDefName("some-task-def-name"); task.setReasonForIncompletion("some-failure-reason"); indexDAO.indexTask(task); await() .atMost(5, TimeUnit.SECONDS) .untilAsserted(() -> { SearchResult<String> result = indexDAO .searchTasks("correlationId='" + correlationId + "'", "*", 0, 10000, null); assertTrue("should return 1 or more search results", result.getResults().size() > 0); assertEquals("taskId should match the indexed task", "some-task-id", result.getResults().get(0)); }); }