ElasticsearchQueryStore implements QueryStore { @Override public void save(String table, Document document) throws QueryStoreException { table = ElasticsearchUtils.getValidTableName(table); try { if (!tableMetadataManager.exists(table)) { throw new QueryStoreException(QueryStoreException.ErrorCode.NO_SUCH_TABLE, "No table exists with the name: " + table); } if (new DateTime().plusDays(1).minus(document.getTimestamp()).getMillis() < 0) { return; } dataStore.save(tableMetadataManager.get(table), document); long timestamp = document.getTimestamp(); Stopwatch stopwatch = new Stopwatch(); stopwatch.start(); connection.getClient() .prepareIndex() .setIndex(ElasticsearchUtils.getCurrentIndex(table, timestamp)) .setType(ElasticsearchUtils.TYPE_NAME) .setId(document.getId()) .setTimestamp(Long.toString(timestamp)) .setSource(mapper.writeValueAsBytes(document.getData())) .setConsistencyLevel(WriteConsistencyLevel.QUORUM) .execute() .get(2, TimeUnit.SECONDS); logger.info(String.format("ES took : %d table : %s", stopwatch.elapsedMillis(), table)); } catch (QueryStoreException ex) { throw ex; } catch (DataStoreException ex) { DataStoreException.ErrorCode code = ex.getErrorCode(); if (code.equals(DataStoreException.ErrorCode.STORE_INVALID_REQUEST) || code.equals(DataStoreException.ErrorCode.STORE_INVALID_DOCUMENT)) { throw new QueryStoreException(QueryStoreException.ErrorCode.INVALID_REQUEST, ex.getMessage(), ex); } else { throw new QueryStoreException(QueryStoreException.ErrorCode.DOCUMENT_SAVE_ERROR, ex.getMessage(), ex); } } catch (JsonProcessingException ex) { throw new QueryStoreException(QueryStoreException.ErrorCode.INVALID_REQUEST, ex.getMessage(), ex); } catch (Exception ex) { throw new QueryStoreException(QueryStoreException.ErrorCode.DOCUMENT_SAVE_ERROR, ex.getMessage(), ex); } } ElasticsearchQueryStore(TableMetadataManager tableMetadataManager,
                                   ElasticsearchConnection connection,
                                   DataStore dataStore); @Override void save(String table, Document document); @Override void save(String table, List<Document> documents); @Override Document get(String table, String id); @Override List<Document> get(String table, List<String> ids); @Override TableFieldMapping getFieldMappings(String table); @Override void cleanupAll(); @Override void cleanup(final String table); @Override void cleanup(Set<String> tables);  }
@Test public void testSaveSingle() throws Exception { Document expectedDocument = new Document(); expectedDocument.setId(UUID.randomUUID().toString()); expectedDocument.setTimestamp(System.currentTimeMillis()); JsonNode data = mapper.valueToTree(Collections.singletonMap("TEST_NAME", "SINGLE_SAVE_TEST")); expectedDocument.setData(data); queryStore.save(TestUtils.TEST_TABLE_NAME, expectedDocument); GetResponse getResponse = elasticsearchServer .getClient() .prepareGet(ElasticsearchUtils.getCurrentIndex(TestUtils.TEST_TABLE_NAME, expectedDocument.getTimestamp()), ElasticsearchUtils.TYPE_NAME, expectedDocument.getId()) .setFields("_timestamp").execute().actionGet(); assertTrue("Id should exist in ES", getResponse.isExists()); assertEquals("Id should match requestId", expectedDocument.getId(), getResponse.getId()); assertEquals("Timestamp should match request timestamp", expectedDocument.getTimestamp(), getResponse.getField("_timestamp").getValue()); }
@Test public void testSaveSingleInvalidTable() throws Exception { Document expectedDocument = new Document(); expectedDocument.setId(UUID.randomUUID().toString()); expectedDocument.setTimestamp(System.currentTimeMillis()); JsonNode data = mapper.valueToTree(Collections.singletonMap("TEST_NAME", "SINGLE_SAVE_TEST")); expectedDocument.setData(data); try { queryStore.save(TestUtils.TEST_TABLE + "-missing", expectedDocument); fail(); } catch (QueryStoreException qse) { assertEquals(QueryStoreException.ErrorCode.NO_SUCH_TABLE, qse.getErrorCode()); } }
@Test public void testSaveBulk() throws Exception { List<Document> documents = new Vector<Document>(); for (int i = 0; i < 10; i++) { documents.add(new Document(UUID.randomUUID().toString(), System.currentTimeMillis(), mapper.valueToTree(Collections.singletonMap("TEST_NAME", "SINGLE_SAVE_TEST")))); } queryStore.save(TestUtils.TEST_TABLE_NAME, documents); for (Document document : documents) { GetResponse getResponse = elasticsearchServer .getClient() .prepareGet(ElasticsearchUtils.getCurrentIndex(TestUtils.TEST_TABLE_NAME, document.getTimestamp()), ElasticsearchUtils.TYPE_NAME, document.getId()) .setFields("_timestamp").execute().actionGet(); assertTrue("Id should exist in ES", getResponse.isExists()); assertEquals("Id should match requestId", document.getId(), getResponse.getId()); assertEquals("Timestamp should match request timestamp", document.getTimestamp(), getResponse.getField("_timestamp").getValue()); } }
@Test public void testSaveBulkNullList() throws Exception { List<Document> list = null; try { queryStore.save(TestUtils.TEST_TABLE_NAME, list); fail(); } catch (QueryStoreException ex) { assertEquals(QueryStoreException.ErrorCode.INVALID_REQUEST, ex.getErrorCode()); } }
@Test public void testSaveBulkEmptyList() throws Exception { List<Document> list = new Vector<Document>(); try { queryStore.save(TestUtils.TEST_TABLE_NAME, list); fail(); } catch (QueryStoreException ex) { assertEquals(QueryStoreException.ErrorCode.INVALID_REQUEST, ex.getErrorCode()); } }
@Test public void testSaveBulkInvalidTable() throws Exception { List<Document> documents = new Vector<Document>(); for (int i = 0; i < 10; i++) { documents.add(new Document(UUID.randomUUID().toString(), System.currentTimeMillis(), mapper.valueToTree(Collections.singletonMap("TEST_NAME", "SINGLE_SAVE_TEST")))); } try { queryStore.save(TestUtils.TEST_TABLE + "-missing", documents); fail(); } catch (QueryStoreException qse) { assertEquals(QueryStoreException.ErrorCode.NO_SUCH_TABLE, qse.getErrorCode()); } }