ElasticSearchDAOV5 implements IndexDAO { @Override public void addMessage(String queue, Message message) { try { long startTime = Instant.now().toEpochMilli(); Map<String, Object> doc = new HashMap<>(); doc.put("messageId", message.getId()); doc.put("payload", message.getPayload()); doc.put("queue", queue); doc.put("created", System.currentTimeMillis()); UpdateRequest req = new UpdateRequest(logIndexName, MSG_DOC_TYPE, message.getId()); req.doc(doc, XContentType.JSON); req.upsert(doc, XContentType.JSON); indexObject(req, MSG_DOC_TYPE); long endTime = Instant.now().toEpochMilli(); logger.debug("Time taken {} for indexing message: {}", endTime - startTime, message.getId()); Monitors.recordESIndexTime("add_message", MSG_DOC_TYPE, endTime - startTime); } catch (Exception e) { logger.error("Failed to index message: {}", message.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 addMessage() { String messageId = "some-message-id"; Message message = new Message(); message.setId(messageId); message.setPayload("some-payload"); message.setReceipt("some-receipt"); indexDAO.addMessage("some-queue", message); await() .atMost(5, TimeUnit.SECONDS) .untilAsserted(() -> { SearchResponse searchResponse = search( LOG_INDEX_PREFIX + "*", "messageId='" + messageId + "'", 0, 10000, "*", MSG_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(), MSG_DOC_TYPE, searchHit.getId()) .get(); assertEquals("indexed message id should match", messageId, response.getSource().get("messageId")); assertEquals("indexed payload should match", "some-payload", response.getSource().get("payload")); }); List<Message> messages = indexDAO.getMessages("some-queue"); assertEquals(1, messages.size()); assertEquals(message.getId(), messages.get(0).getId()); assertEquals(message.getPayload(), messages.get(0).getPayload()); }