From 8eee517a493c42d2f26d71ccd8b14a4cf26d24a7 Mon Sep 17 00:00:00 2001 From: Corneil du Plessis Date: Thu, 1 Dec 2022 15:17:44 +0200 Subject: [PATCH] Update test-binder dependency. Added elasticsearch-x-content. --- applications/sink/pgcopy-sink/pom.xml | 4 +-- applications/sink/rabbit-sink/pom.xml | 6 ++-- applications/source/jdbc-source/pom.xml | 4 +-- applications/stream-applications-core/pom.xml | 4 +-- .../pom.xml | 4 +-- .../consumer/elasticsearch-consumer/pom.xml | 6 ++++ .../ElasticsearchConsumerConfiguration.java | 5 +-- ...ElasticsearchConsumerApplicationTests.java | 34 +++++++++++++++++++ 8 files changed, 49 insertions(+), 18 deletions(-) diff --git a/applications/sink/pgcopy-sink/pom.xml b/applications/sink/pgcopy-sink/pom.xml index 4ed38b82..c4280afd 100644 --- a/applications/sink/pgcopy-sink/pom.xml +++ b/applications/sink/pgcopy-sink/pom.xml @@ -30,10 +30,8 @@ org.springframework.cloud - spring-cloud-stream - test-jar + spring-cloud-stream-test-binder test - test-binder diff --git a/applications/sink/rabbit-sink/pom.xml b/applications/sink/rabbit-sink/pom.xml index 220eb57a..cc79026c 100644 --- a/applications/sink/rabbit-sink/pom.xml +++ b/applications/sink/rabbit-sink/pom.xml @@ -31,10 +31,8 @@ org.springframework.cloud - spring-cloud-stream - test-jar - test - test-binder + spring-cloud-stream-test-binder + test org.testcontainers diff --git a/applications/source/jdbc-source/pom.xml b/applications/source/jdbc-source/pom.xml index ba005e5f..652b878f 100644 --- a/applications/source/jdbc-source/pom.xml +++ b/applications/source/jdbc-source/pom.xml @@ -21,10 +21,8 @@ org.springframework.cloud - spring-cloud-stream - test-jar + spring-cloud-stream-test-binder test - test-binder com.h2database diff --git a/applications/stream-applications-core/pom.xml b/applications/stream-applications-core/pom.xml index 7745734c..d60cbd72 100644 --- a/applications/stream-applications-core/pom.xml +++ b/applications/stream-applications-core/pom.xml @@ -81,9 +81,7 @@ org.springframework.cloud - spring-cloud-stream - test-jar - test-binder + spring-cloud-stream-test-binder test diff --git a/applications/stream-applications-core/stream-applications-security-common/pom.xml b/applications/stream-applications-core/stream-applications-security-common/pom.xml index 4ef83b96..ee5ae147 100644 --- a/applications/stream-applications-core/stream-applications-security-common/pom.xml +++ b/applications/stream-applications-core/stream-applications-security-common/pom.xml @@ -51,9 +51,7 @@ org.springframework.cloud - spring-cloud-stream - test-jar - test-binder + spring-cloud-stream-test-binder test diff --git a/functions/consumer/elasticsearch-consumer/pom.xml b/functions/consumer/elasticsearch-consumer/pom.xml index a38b0c40..fb00f8af 100644 --- a/functions/consumer/elasticsearch-consumer/pom.xml +++ b/functions/consumer/elasticsearch-consumer/pom.xml @@ -23,6 +23,12 @@ elasticsearch-java ${elasticsearch.version} + + org.elasticsearch + elasticsearch-x-content + ${elasticsearch.version} + + org.springframework.boot spring-boot-starter-data-elasticsearch diff --git a/functions/consumer/elasticsearch-consumer/src/main/java/org/springframework/cloud/fn/consumer/elasticsearch/ElasticsearchConsumerConfiguration.java b/functions/consumer/elasticsearch-consumer/src/main/java/org/springframework/cloud/fn/consumer/elasticsearch/ElasticsearchConsumerConfiguration.java index ea382fc3..b60ba662 100644 --- a/functions/consumer/elasticsearch-consumer/src/main/java/org/springframework/cloud/fn/consumer/elasticsearch/ElasticsearchConsumerConfiguration.java +++ b/functions/consumer/elasticsearch-consumer/src/main/java/org/springframework/cloud/fn/consumer/elasticsearch/ElasticsearchConsumerConfiguration.java @@ -60,6 +60,7 @@ import org.springframework.util.StringUtils; /** * @author Soby Chacko * @author Andrea Montemaggio + * @author Corneil du Plessis */ @Configuration @EnableConfigurationProperties(ElasticsearchConsumerProperties.class) @@ -179,7 +180,7 @@ public class ElasticsearchConsumerConfiguration { requestBuilder.withJson(new StringReader((String) message.getPayload())); } else if (message.getPayload() instanceof Map) { - requestBuilder.document((Map) message.getPayload()); + requestBuilder.document(message.getPayload()); } if (StringUtils.hasText(consumerProperties.getRouting())) { @@ -274,7 +275,7 @@ public class ElasticsearchConsumerConfiguration { .stream() .map(bulkResponseItem -> bulkResponseItem.error() != null ? bulkResponseItem.error().toString() : "") .reduce((errorCause, errorCause2) -> errorCause != null ? errorCause + " : " + errorCause2 : errorCause2) - .orElseGet(() -> response.toString()); + .orElseGet(response::toString); throw new IllegalStateException("Bulk indexing operation completed with failures: " + error); } } diff --git a/functions/consumer/elasticsearch-consumer/src/test/java/org/springframework/cloud/fn/consumer/elasticsearch/ElasticsearchConsumerApplicationTests.java b/functions/consumer/elasticsearch-consumer/src/test/java/org/springframework/cloud/fn/consumer/elasticsearch/ElasticsearchConsumerApplicationTests.java index 44be210c..cea3c7d0 100644 --- a/functions/consumer/elasticsearch-consumer/src/test/java/org/springframework/cloud/fn/consumer/elasticsearch/ElasticsearchConsumerApplicationTests.java +++ b/functions/consumer/elasticsearch-consumer/src/test/java/org/springframework/cloud/fn/consumer/elasticsearch/ElasticsearchConsumerApplicationTests.java @@ -28,6 +28,9 @@ import co.elastic.clients.elasticsearch.core.GetRequest; import co.elastic.clients.elasticsearch.core.GetResponse; import co.elastic.clients.json.JsonData; import org.awaitility.Awaitility; +import org.elasticsearch.xcontent.XContentBuilder; +import org.elasticsearch.xcontent.XContentFactory; +import org.junit.jupiter.api.Disabled; import org.junit.jupiter.api.Tag; import org.junit.jupiter.api.Test; import org.testcontainers.elasticsearch.ElasticsearchContainer; @@ -49,6 +52,8 @@ import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThatExceptionOfType; import static org.assertj.core.api.Assertions.assertThatIllegalStateException; +// import static org.elasticsearch.core.Strings.*; + /** * @author Soby Chacko * @author Andrea Montemaggio @@ -146,6 +151,35 @@ public class ElasticsearchConsumerApplicationTests { }); } + @Test + @Disabled + public void testXContentBuilder() { + this.contextRunner + .withPropertyValues("elasticsearch.consumer.index=foo", "elasticsearch.consumer.id=4", + "spring.elasticsearch.rest.uris=http://" + elasticsearch.getHttpHostAddress()) + .run(context -> { + Consumer> elasticsearchConsumer = context.getBean("elasticsearchConsumer", Consumer.class); + + XContentBuilder builder = XContentFactory.jsonBuilder(); + builder.startObject(); + builder.field("user", "kimchy"); + builder.timeField("postDate", 1471466076564L); + builder.field("message", "trying out Elasticsearch"); + builder.endObject(); + + final Message message = MessageBuilder.withPayload(builder).build(); + + elasticsearchConsumer.accept(message); + + final ElasticsearchClient elasticsearchClient = context.getBean(ElasticsearchClient.class); + final GetRequest getRequest = new GetRequest.Builder().index("foo").id("4").build(); + + final GetResponse response = elasticsearchClient.get(getRequest, String.class); + assertThat(response.found()).isTrue(); + + assertThat(response.source()).isEqualTo(builder.toString()); + }); + } @Test public void testAsyncIndexing() { this.contextRunner