ElasticsearchClient { void sendBulk(HttpClient.OutputStreamHandler outputStreamHandler) { sendBulk(outputStreamHandler, true); } ElasticsearchClient(final CorePlugin corePlugin, final HttpClient httpClient, int esAvailabilityCheckIntervalSec, List<ElasticsearchAvailabilityObserver> elasticsearchAvailabilityObservers); boolean isElasticsearch7Compatible(); boolean isElasticsearch6Compatible(); String getElasticSearchTemplateResourcePath(); String getKibanaResourcePath(); JsonNode getJson(final String path); JsonNode getJson(final String path, final boolean suppressWarning); T getObject(final String path, Class<T> type); Collection<T> getAll(String path, int limit, Class<T> clazz); int delete(final String path); int sendAsJson(final String method, final String path, final Object requestBody); void updateKibanaIndexPattern(final String indexName, final String indexPatternLocation); void sendMappingTemplate(String mappingJson, String templateName); static String modifyIndexTemplate(String templatePath, int moveToColdNodesAfterDays, Integer numberOfReplicas, Integer numberOfShards); void sendMetricDashboardBulkAsync(final String resource); void sendSpanDashboardBulkAsync(final String resource, boolean logBulkErrors); void deleteIndices(String indexPattern); void forceMergeIndices(String indexPattern); void updateIndexSettings(String indexPattern, final Map<String, ?> settings); void waitForCompletion(); void close(); void scheduleIndexManagement(String indexPrefix, int optimizeAndMoveIndicesToColdNodesOlderThanDays, int deleteIndicesOlderThanDays); static String getBulkHeader(String action, String index, String type); boolean isElasticsearchAvailable(); HttpClient getHttpClient(); URL getElasticsearchUrl(); void createIndexAndSendMapping(final String index, final String type, final InputStream mapping); void createEmptyIndex(String indexName); void checkEsAvailability(); static final Map<String, String> CONTENT_TYPE_NDJSON; }
@Test public void testBulkNoRequest() { elasticsearchClient.sendBulk(os -> os.write(("").getBytes("UTF-8"))); assertThat(testAppender.list).hasSize(1); final ILoggingEvent event = testAppender.list.get(0); assertThat(event.getLevel().toString()).isEqualTo("WARN"); assertThat(event.getMessage()).startsWith("Error(s) while sending a _bulk request to elasticsearch: {}"); }
@Test public void testBulkErrorInvalidRequest() { elasticsearchClient.sendBulk(os -> os.write(("{ \"update\" : {\"_id\" : \"1\", \"_type\" : \"type1\", \"_index\" : \"index1\", \"_retry_on_conflict\" : 3} }\n" + "{ \"doc\" : {\"field\" : \"value\"} }\n").getBytes("UTF-8"))); assertThat(testAppender.list).hasSize(1); final ILoggingEvent event = testAppender.list.get(0); assertThat(event.getLevel().toString()).isEqualTo("WARN"); assertThat(event.getMessage()).startsWith("Error(s) while sending a _bulk request to elasticsearch: {}"); }
@Test public void testSuccessfulBulkRequest() { elasticsearchClient.sendBulk(os -> os.write(("{ \"index\" : { \"_index\" : \"test\", \"_type\" : \"type1\", \"_id\" : \"1\" } }\n" + "{ \"field1\" : \"value1\" }\n").getBytes("UTF-8"))); assertThat(testAppender.list).hasSize(0); }