Update test-binder dependency.

Added elasticsearch-x-content.

applications/sink/pgcopy-sink/pom.xml

@@ -30,10 +30,8 @@
 		</dependency>
 		<dependency>
 			<groupId>org.springframework.cloud</groupId>
-			<artifactId>spring-cloud-stream</artifactId>
-			<type>test-jar</type>
+			<artifactId>spring-cloud-stream-test-binder</artifactId>
 			<scope>test</scope>
-			<classifier>test-binder</classifier>
 		</dependency>
 	</dependencies>
 

applications/sink/rabbit-sink/pom.xml

@@ -31,10 +31,8 @@
         </dependency>
 		<dependency>
 			<groupId>org.springframework.cloud</groupId>
-			<artifactId>spring-cloud-stream</artifactId>
-			<type>test-jar</type>
-			<scope>test</scope>
-			<classifier>test-binder</classifier>
+            <artifactId>spring-cloud-stream-test-binder</artifactId>
+            <scope>test</scope>
 		</dependency>
         <dependency>
             <groupId>org.testcontainers</groupId>

applications/source/jdbc-source/pom.xml

@@ -21,10 +21,8 @@
         </dependency>
 		<dependency>
 			<groupId>org.springframework.cloud</groupId>
-			<artifactId>spring-cloud-stream</artifactId>
-			<type>test-jar</type>
+			<artifactId>spring-cloud-stream-test-binder</artifactId>
 			<scope>test</scope>
-			<classifier>test-binder</classifier>
 		</dependency>
         <dependency>
             <groupId>com.h2database</groupId>

applications/stream-applications-core/pom.xml

@@ -81,9 +81,7 @@
         </dependency>
         <dependency>
             <groupId>org.springframework.cloud</groupId>
-            <artifactId>spring-cloud-stream</artifactId>
-            <type>test-jar</type>
-            <classifier>test-binder</classifier>
+            <artifactId>spring-cloud-stream-test-binder</artifactId>
             <scope>test</scope>
         </dependency>
     </dependencies>

applications/stream-applications-core/stream-applications-security-common/pom.xml

@@ -51,9 +51,7 @@
         </dependency>
         <dependency>
             <groupId>org.springframework.cloud</groupId>
-            <artifactId>spring-cloud-stream</artifactId>
-            <type>test-jar</type>
-            <classifier>test-binder</classifier>
+            <artifactId>spring-cloud-stream-test-binder</artifactId>
             <scope>test</scope>
         </dependency>
     </dependencies>

functions/consumer/elasticsearch-consumer/pom.xml

@@ -23,6 +23,12 @@
             <artifactId>elasticsearch-java</artifactId>
             <version>${elasticsearch.version}</version>
         </dependency>
+        <dependency>
+            <groupId>org.elasticsearch</groupId>
+            <artifactId>elasticsearch-x-content</artifactId>
+            <version>${elasticsearch.version}</version>
+        </dependency>
+
         <dependency>
             <groupId>org.springframework.boot</groupId>
             <artifactId>spring-boot-starter-data-elasticsearch</artifactId>

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<String, ?>) 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);
 		}
 	}

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<Message<?>> 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<XContentBuilder> 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<String> response = elasticsearchClient.get(getRequest, String.class);
+				assertThat(response.found()).isTrue();
+
+				assertThat(response.source()).isEqualTo(builder.toString());
+			});
+	}
 	@Test
 	public void testAsyncIndexing() {
 		this.contextRunner