ElasticSearchDAOV5 implements IndexDAO { @Override public void addEventExecution(EventExecution eventExecution) { try { long startTime = Instant.now().toEpochMilli(); byte[] doc = objectMapper.writeValueAsBytes(eventExecution); String id = eventExecution.getName() + "." + eventExecution.getEvent() + "." + eventExecution .getMessageId() + "." + eventExecution.getId(); UpdateRequest req = new UpdateRequest(logIndexName, EVENT_DOC_TYPE, id); req.doc(doc, XContentType.JSON); req.upsert(doc, XContentType.JSON); indexObject(req, EVENT_DOC_TYPE); long endTime = Instant.now().toEpochMilli(); logger.debug("Time taken {} for indexing event execution: {}", endTime - startTime, eventExecution.getId()); Monitors.recordESIndexTime("add_event_execution", EVENT_DOC_TYPE, endTime - startTime); Monitors.recordWorkerQueueSize("logQueue", ((ThreadPoolExecutor) logExecutorService).getQueue().size()); } catch (Exception e) { logger.error("Failed to index event execution: {}", eventExecution.getId(), e); } } @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 addEventExecution() { String messageId = "some-message-id"; EventExecution eventExecution = new EventExecution(); eventExecution.setId("some-id"); eventExecution.setMessageId(messageId); eventExecution.setAction(Type.complete_task); eventExecution.setEvent("some-event"); eventExecution.setStatus(EventExecution.Status.COMPLETED); indexDAO.addEventExecution(eventExecution); await() .atMost(5, TimeUnit.SECONDS) .untilAsserted(() -> { SearchResponse searchResponse = search( LOG_INDEX_PREFIX + "*", "messageId='" + messageId + "'", 0, 10000, "*", EVENT_DOC_TYPE ); assertEquals("search results should be length 1", searchResponse.getHits().getTotalHits(), 1); SearchHit searchHit = searchResponse.getHits().getAt(0); GetResponse response = elasticSearchClient .prepareGet(searchHit.getIndex(), EVENT_DOC_TYPE, searchHit.getId()) .get(); assertEquals("indexed message id should match", messageId, response.getSource().get("messageId")); assertEquals("indexed id should match", "some-id", response.getSource().get("id")); assertEquals("indexed status should match", EventExecution.Status.COMPLETED.name(), response.getSource().get("status")); }); List<EventExecution> events = indexDAO.getEventExecutions("some-event"); assertEquals(1, events.size()); assertEquals(eventExecution, events.get(0)); }