From 93fdd2ef0f5b4fb4127afa9db571ab66fdc5b473 Mon Sep 17 00:00:00 2001
From: Gary Russell <grussell@pivotal.io>
Date: Sat, 30 Sep 2017 10:06:00 -0400
Subject: [PATCH] Update to SK 2.0.1.BUILD-SNAPSHOT

---
 pom.xml                                       |  2 +-
 spring-cloud-stream-binder-kafka-test/pom.xml |  2 +-
 .../stream/binder/kafka/KafkaBinderTests.java | 41 +++++++++++++++++++
 3 files changed, 43 insertions(+), 2 deletions(-)

diff --git a/pom.xml b/pom.xml
index 7518335ea..5833cbee4 100644
--- a/pom.xml
+++ b/pom.xml
@@ -12,7 +12,7 @@
 	</parent>
 	<properties>
 		<java.version>1.8</java.version>
-		<spring-kafka.version>2.0.0.BUILD-SNAPSHOT</spring-kafka.version>
+		<spring-kafka.version>2.0.1.BUILD-SNAPSHOT</spring-kafka.version>
 		<kafka.version>0.11.0.0</kafka.version>
 		<spring-integration-kafka.version>3.0.0.BUILD-SNAPSHOT</spring-integration-kafka.version>
 		<spring-cloud-stream.version>2.0.0.BUILD-SNAPSHOT</spring-cloud-stream.version>
diff --git a/spring-cloud-stream-binder-kafka-test/pom.xml b/spring-cloud-stream-binder-kafka-test/pom.xml
index d1da44205..798f21be3 100644
--- a/spring-cloud-stream-binder-kafka-test/pom.xml
+++ b/spring-cloud-stream-binder-kafka-test/pom.xml
@@ -19,7 +19,7 @@
     <properties>
         <main.basedir>${basedir}/../..</main.basedir>
         <kafka.version>0.11.0.0</kafka.version>
-        <spring-kafka.version>2.0.0.BUILD-SNAPSHOT</spring-kafka.version>
+        <spring-kafka.version>2.0.1.BUILD-SNAPSHOT</spring-kafka.version>
     </properties>
 
     <dependencies>
diff --git a/spring-cloud-stream-binder-kafka-test/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java b/spring-cloud-stream-binder-kafka-test/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java
index 05c4c893e..07063b150 100644
--- a/spring-cloud-stream-binder-kafka-test/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java
+++ b/spring-cloud-stream-binder-kafka-test/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java
@@ -25,8 +25,10 @@ import java.util.UUID;
 
 import io.confluent.kafka.schemaregistry.rest.SchemaRegistryConfig;
 import io.confluent.kafka.schemaregistry.rest.SchemaRegistryRestApplication;
+
 import kafka.utils.ZKStringSerializer$;
 import kafka.utils.ZkUtils;
+
 import org.I0Itec.zkclient.ZkClient;
 import org.apache.kafka.clients.consumer.ConsumerConfig;
 import org.apache.kafka.common.serialization.ByteArrayDeserializer;
@@ -38,6 +40,7 @@ import org.junit.ClassRule;
 import org.junit.Test;
 
 import org.springframework.cloud.stream.binder.Binder;
+import org.springframework.cloud.stream.binder.BinderHeaders;
 import org.springframework.cloud.stream.binder.Binding;
 import org.springframework.cloud.stream.binder.ExtendedConsumerProperties;
 import org.springframework.cloud.stream.binder.ExtendedProducerProperties;
@@ -59,6 +62,7 @@ import org.springframework.messaging.MessageChannel;
 import org.springframework.messaging.MessageHeaders;
 import org.springframework.messaging.SubscribableChannel;
 import org.springframework.messaging.support.MessageBuilder;
+import org.springframework.util.MimeType;
 import org.springframework.util.MimeTypeUtils;
 
 import static org.assertj.core.api.Assertions.assertThat;
@@ -183,6 +187,43 @@ public class KafkaBinderTests extends AbstractKafkaBinderTests {
 		return new DefaultKafkaConsumerFactory<>(props, keyDecoder, valueDecoder);
 	}
 
+	@SuppressWarnings({ "rawtypes", "unchecked" })
+	@Test
+	public void testTrustedPackages() throws Exception {
+		Binder binder = getBinder();
+
+		BindingProperties producerBindingProperties = createProducerBindingProperties(createProducerProperties());
+		DirectChannel moduleOutputChannel = createBindableChannel("output", producerBindingProperties);
+		QueueChannel moduleInputChannel = new QueueChannel();
+		Binding<MessageChannel> producerBinding = binder.bindProducer("bar.0", moduleOutputChannel,
+				producerBindingProperties.getProducer());
+		ExtendedConsumerProperties<KafkaConsumerProperties> consumerProperties = createConsumerProperties();
+		consumerProperties.getExtension().setTrustedPackages(new String[]{"org.springframework.util"});
+		Binding<MessageChannel> consumerBinding = binder.bindConsumer("bar.0",
+				"testSendAndReceiveNoOriginalContentType", moduleInputChannel, consumerProperties);
+		binderBindUnbindLatency();
+
+		Message<?> message = org.springframework.integration.support.MessageBuilder.withPayload("foo")
+			.setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.TEXT_PLAIN_VALUE)
+			.setHeader("foo", MimeTypeUtils.TEXT_PLAIN)
+			.build();
+
+		moduleOutputChannel.send(message);
+		Message<?> inbound = receive(moduleInputChannel);
+		Assertions.assertThat(inbound).isNotNull();
+		Assertions.assertThat(inbound.getPayload()).isEqualTo("foo".getBytes());
+		Assertions.assertThat(inbound.getHeaders().get(BinderHeaders.BINDER_ORIGINAL_CONTENT_TYPE)).isNull();
+		Assertions.assertThat(inbound.getHeaders().get(MessageHeaders.CONTENT_TYPE))
+				.isEqualTo(MimeTypeUtils.TEXT_PLAIN);
+		Assertions.assertThat(inbound.getHeaders().get("foo")).isInstanceOf(MimeType.class);
+		MimeType actual = (MimeType) inbound.getHeaders().get("foo");
+		Assertions.assertThat(actual).isEqualTo(MimeTypeUtils.TEXT_PLAIN);
+		producerBinding.unbind();
+		consumerBinding.unbind();
+	}
+
+	class Foo{}
+
 	@Test
 	@SuppressWarnings("unchecked")
 	public void testCustomAvroSerialization() throws Exception {