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