@Test public void storeItems() { final Collection<QueryVersionReport.VersionMetric> emptyMetrics = Collections.emptyList(); final Collection<QueryVersionReport.Result> emptyResult = Collections.emptyList(); final QueryVersionReport qvr1 = new QueryVersionReport("1", null, null, null, null, "1.0", 0, emptyMetrics, emptyResult); final QueryVersionReport qvr2 = new QueryVersionReport("2", null, null, null, null, "1.1", 0, emptyMetrics, emptyResult); mockServerClient.when(request() .withPath("/_bulk") .withMethod(HttpPost.METHOD_NAME)) .respond(response() .withBody("{\"took\":63,\"errors\":false,\"items\":" + "[{\"index\":{\"_index\":\"rre\",\"_type\":\"_doc\",\"_id\":\"1\",\"_version\":1,\"result\":\"created\",\"_shards\":{\"total\":2,\"successful\":1,\"failed\":0},\"_seq_no\":0,\"_primary_term\":4,\"status\":201}}," + "{\"index\":{\"_index\":\"rre\",\"_type\":\"_doc\",\"_id\":\"2\",\"_version\":1,\"result\":\"created\",\"_shards\":{\"total\":2,\"successful\":1,\"failed\":0},\"_seq_no\":0,\"_primary_term\":4,\"status\":201}}]}")); connector.storeItems(INDEX_NAME, Arrays.asList(qvr1, qvr2)); mockServerClient.verify( request() .withMethod(HttpPost.METHOD_NAME) .withPath("/_bulk") .withBody("{\"index\":{\"_index\":\"rre\",\"_id\":\"1\"}}\n" + "{\"id\":\"1\",\"version\":\"1.0\",\"totalHits\":0}\n" + "{\"index\":{\"_index\":\"rre\",\"_id\":\"2\"}}\n" + "{\"id\":\"2\",\"version\":\"1.1\",\"totalHits\":0}\n")); }
@Override public void storeItems(String index, Collection<QueryVersionReport> reports) { BulkProcessor.Listener listener = new BulkProcessor.Listener() { @Override public void beforeBulk(long l, BulkRequest bulkRequest) { LOGGER.debug("About to execute bulk request of {} actions", bulkRequest.numberOfActions()); } @Override public void afterBulk(long l, BulkRequest bulkRequest, BulkResponse bulkResponse) { if (bulkResponse.hasFailures()) { LOGGER.warn("Bulk update request had failures!"); LOGGER.warn(bulkResponse.buildFailureMessage()); } } @Override public void afterBulk(long l, BulkRequest bulkRequest, Throwable throwable) { LOGGER.error("Caught exception while executing bulk request: " + throwable.getMessage()); } }; BulkProcessor processor = BulkProcessor.builder((bulkRequest, bulkListener) -> client.bulkAsync(bulkRequest, RequestOptions.DEFAULT, bulkListener), listener) .build(); final ObjectMapper mapper = new ObjectMapper(); reports.forEach(r -> processor.add( new IndexRequest(index) .id(r.getId()) .source(ConnectorUtils.convertReportToJson(mapper, r), XContentType.JSON))); processor.flush(); try { if (!processor.awaitClose(30, TimeUnit.SECONDS)) { LOGGER.warn("Bulk update processor was terminated before it could complete operations!"); } } catch (InterruptedException e) { LOGGER.error("Bulk update processor was interrupted before it could be closed :: {}", e.getMessage()); } }
IndexOnlyElasticsearchConnector implements ElasticsearchConnector { @Override public void storeItems(String index, Collection<QueryVersionReport> reports) { BulkProcessor.Listener listener = new BulkProcessor.Listener() { @Override public void beforeBulk(long l, BulkRequest bulkRequest) { LOGGER.debug("About to execute bulk request of {} actions", bulkRequest.numberOfActions()); } @Override public void afterBulk(long l, BulkRequest bulkRequest, BulkResponse bulkResponse) { if (bulkResponse.hasFailures()) { LOGGER.warn("Bulk update request had failures!"); LOGGER.warn(bulkResponse.buildFailureMessage()); } } @Override public void afterBulk(long l, BulkRequest bulkRequest, Throwable throwable) { LOGGER.error("Caught exception while executing bulk request: " + throwable.getMessage()); } }; BulkProcessor processor = BulkProcessor.builder((bulkRequest, bulkListener) -> client.bulkAsync(bulkRequest, RequestOptions.DEFAULT, bulkListener), listener) .build(); final ObjectMapper mapper = new ObjectMapper(); reports.forEach(r -> processor.add( new IndexRequest(index) .id(r.getId()) .source(ConnectorUtils.convertReportToJson(mapper, r), XContentType.JSON))); processor.flush(); try { if (!processor.awaitClose(30, TimeUnit.SECONDS)) { LOGGER.warn("Bulk update processor was terminated before it could complete operations!"); } } catch (InterruptedException e) { LOGGER.error("Bulk update processor was interrupted before it could be closed :: {}", e.getMessage()); } } }
IndexOnlyElasticsearchConnector implements ElasticsearchConnector { @Override public void storeItems(String index, Collection<QueryVersionReport> reports) { BulkProcessor.Listener listener = new BulkProcessor.Listener() { @Override public void beforeBulk(long l, BulkRequest bulkRequest) { LOGGER.debug("About to execute bulk request of {} actions", bulkRequest.numberOfActions()); } @Override public void afterBulk(long l, BulkRequest bulkRequest, BulkResponse bulkResponse) { if (bulkResponse.hasFailures()) { LOGGER.warn("Bulk update request had failures!"); LOGGER.warn(bulkResponse.buildFailureMessage()); } } @Override public void afterBulk(long l, BulkRequest bulkRequest, Throwable throwable) { LOGGER.error("Caught exception while executing bulk request: " + throwable.getMessage()); } }; BulkProcessor processor = BulkProcessor.builder((bulkRequest, bulkListener) -> client.bulkAsync(bulkRequest, RequestOptions.DEFAULT, bulkListener), listener) .build(); final ObjectMapper mapper = new ObjectMapper(); reports.forEach(r -> processor.add( new IndexRequest(index) .id(r.getId()) .source(ConnectorUtils.convertReportToJson(mapper, r), XContentType.JSON))); processor.flush(); try { if (!processor.awaitClose(30, TimeUnit.SECONDS)) { LOGGER.warn("Bulk update processor was terminated before it could complete operations!"); } } catch (InterruptedException e) { LOGGER.error("Bulk update processor was interrupted before it could be closed :: {}", e.getMessage()); } } IndexOnlyElasticsearchConnector(RestHighLevelClient client); }
IndexOnlyElasticsearchConnector implements ElasticsearchConnector { @Override public void storeItems(String index, Collection<QueryVersionReport> reports) { BulkProcessor.Listener listener = new BulkProcessor.Listener() { @Override public void beforeBulk(long l, BulkRequest bulkRequest) { LOGGER.debug("About to execute bulk request of {} actions", bulkRequest.numberOfActions()); } @Override public void afterBulk(long l, BulkRequest bulkRequest, BulkResponse bulkResponse) { if (bulkResponse.hasFailures()) { LOGGER.warn("Bulk update request had failures!"); LOGGER.warn(bulkResponse.buildFailureMessage()); } } @Override public void afterBulk(long l, BulkRequest bulkRequest, Throwable throwable) { LOGGER.error("Caught exception while executing bulk request: " + throwable.getMessage()); } }; BulkProcessor processor = BulkProcessor.builder((bulkRequest, bulkListener) -> client.bulkAsync(bulkRequest, RequestOptions.DEFAULT, bulkListener), listener) .build(); final ObjectMapper mapper = new ObjectMapper(); reports.forEach(r -> processor.add( new IndexRequest(index) .id(r.getId()) .source(ConnectorUtils.convertReportToJson(mapper, r), XContentType.JSON))); processor.flush(); try { if (!processor.awaitClose(30, TimeUnit.SECONDS)) { LOGGER.warn("Bulk update processor was terminated before it could complete operations!"); } } catch (InterruptedException e) { LOGGER.error("Bulk update processor was interrupted before it could be closed :: {}", e.getMessage()); } } IndexOnlyElasticsearchConnector(RestHighLevelClient client); @Override boolean isAvailable(); @Override boolean indexExists(String index); @Override boolean createIndex(String index); @Override void storeItems(String index, Collection<QueryVersionReport> reports); @Override void close(); }
IndexOnlyElasticsearchConnector implements ElasticsearchConnector { @Override public void storeItems(String index, Collection<QueryVersionReport> reports) { BulkProcessor.Listener listener = new BulkProcessor.Listener() { @Override public void beforeBulk(long l, BulkRequest bulkRequest) { LOGGER.debug("About to execute bulk request of {} actions", bulkRequest.numberOfActions()); } @Override public void afterBulk(long l, BulkRequest bulkRequest, BulkResponse bulkResponse) { if (bulkResponse.hasFailures()) { LOGGER.warn("Bulk update request had failures!"); LOGGER.warn(bulkResponse.buildFailureMessage()); } } @Override public void afterBulk(long l, BulkRequest bulkRequest, Throwable throwable) { LOGGER.error("Caught exception while executing bulk request: " + throwable.getMessage()); } }; BulkProcessor processor = BulkProcessor.builder((bulkRequest, bulkListener) -> client.bulkAsync(bulkRequest, RequestOptions.DEFAULT, bulkListener), listener) .build(); final ObjectMapper mapper = new ObjectMapper(); reports.forEach(r -> processor.add( new IndexRequest(index) .id(r.getId()) .source(ConnectorUtils.convertReportToJson(mapper, r), XContentType.JSON))); processor.flush(); try { if (!processor.awaitClose(30, TimeUnit.SECONDS)) { LOGGER.warn("Bulk update processor was terminated before it could complete operations!"); } } catch (InterruptedException e) { LOGGER.error("Bulk update processor was interrupted before it could be closed :: {}", e.getMessage()); } } IndexOnlyElasticsearchConnector(RestHighLevelClient client); @Override boolean isAvailable(); @Override boolean indexExists(String index); @Override boolean createIndex(String index); @Override void storeItems(String index, Collection<QueryVersionReport> reports); @Override void close();  }