ElasticsearchQueryStore implements QueryStore { @Override @Timed public void save(String table, Document document) { table = ElasticsearchUtils.getValidTableName(table); Stopwatch stopwatch = Stopwatch.createStarted(); String action = StringUtils.EMPTY; try { if(!tableMetadataManager.exists(table)) { throw FoxtrotExceptions.createBadRequestException(table, String.format(UNKNOWN_TABLE_ERROR_MESSAGE, table)); } if(new DateTime().plusDays(1) .minus(document.getTimestamp()) .getMillis() < 0) { return; } action = TABLE_META; stopwatch.reset() .start(); final Table tableMeta = tableMetadataManager.get(table); logger.debug("TableMetaGetTook:{}", stopwatch.elapsed(TimeUnit.MILLISECONDS)); MetricUtil.getInstance() .registerActionSuccess(action, table, stopwatch.elapsed(TimeUnit.MILLISECONDS)); stopwatch.reset() .start(); action = DATA_STORE; final Document translatedDocument = dataStore.save(tableMeta, document); logger.debug("DataStoreTook:{}", stopwatch.elapsed(TimeUnit.MILLISECONDS)); MetricUtil.getInstance() .registerActionSuccess(action, table, stopwatch.elapsed(TimeUnit.MILLISECONDS)); stopwatch.reset() .start(); action = QUERY_STORE; long timestamp = translatedDocument.getTimestamp(); connection.getClient() .prepareIndex() .setIndex(ElasticsearchUtils.getCurrentIndex(table, timestamp)) .setType(ElasticsearchUtils.DOCUMENT_TYPE_NAME) .setId(translatedDocument.getId()) .setSource(convert(table, translatedDocument)) .execute() .get(2, TimeUnit.SECONDS); logger.debug("QueryStoreTook:{}", stopwatch.elapsed(TimeUnit.MILLISECONDS)); MetricUtil.getInstance() .registerActionSuccess(action, table, stopwatch.elapsed(TimeUnit.MILLISECONDS)); } catch (InterruptedException | ExecutionException | TimeoutException e) { MetricUtil.getInstance() .registerActionFailure(action, table, stopwatch.elapsed(TimeUnit.MILLISECONDS)); Thread.currentThread() .interrupt(); throw FoxtrotExceptions.createExecutionException(table, e); } } @Inject  ElasticsearchQueryStore(TableMetadataManager tableMetadataManager,
                                   ElasticsearchConnection connection,
                                   DataStore dataStore,
                                   List<IndexerEventMutator> mutators,
                                   ObjectMapper mapper,
                                   CardinalityConfig cardinalityConfig); @Override @Timed void initializeTable(String table); @Override @Timed void save(String table, Document document); @Override @Timed void save(String table, List<Document> documents); @Override @Timed Document get(String table, String id); @Override List<Document> getAll(String table, List<String> ids); @Override @Timed List<Document> getAll(String table, List<String> ids, boolean bypassMetalookup); @Override void cleanupAll(); @Override @Timed void cleanup(final String table); @Override @Timed void cleanup(Set<String> tables); @Override ClusterHealthResponse getClusterHealth(); @Override NodesStatsResponse getNodeStats(); @Override IndicesStatsResponse getIndicesStats(); @Override TableFieldMapping getFieldMappings(String table);  }
@Test public void testSaveSingleInvalidTable() throws Exception { Document expectedDocument = createDummyDocument(); try { queryStore.save(TestUtils.TEST_TABLE + "-missing", expectedDocument); fail(); } catch (FoxtrotException qse) { assertEquals(ErrorCode.INVALID_REQUEST, qse.getCode()); } }
@Test public void testSaveBulkNullList() throws Exception { List<Document> list = null; try { queryStore.save(TestUtils.TEST_TABLE_NAME, list); fail(); } catch (FoxtrotException ex) { assertEquals(ErrorCode.INVALID_REQUEST, ex.getCode()); } }
@Test public void testSaveBulkEmptyList() throws Exception { List<Document> list = new Vector<Document>(); try { queryStore.save(TestUtils.TEST_TABLE_NAME, list); fail(); } catch (FoxtrotException ex) { assertEquals(ErrorCode.INVALID_REQUEST, ex.getCode()); } }
@Test public void testSaveBulkInvalidTable() throws Exception { List<Document> documents = new Vector<Document>(); for (int i = 0; i < 10; i++) { documents.add(createDummyDocument()); } try { queryStore.save(TestUtils.TEST_TABLE + "-missing", documents); fail(); } catch (FoxtrotException qse) { assertEquals(ErrorCode.INVALID_REQUEST, qse.getCode()); } }