From c320ba45191a1a44c13a404492f0e11169f34f47 Mon Sep 17 00:00:00 2001
From: Gustavo Martin Morcuende <gu.martinm@gmail.com>
Date: Sun, 29 Jan 2017 10:55:07 +0100
Subject: [PATCH] Using Kafka 0.10.1.1

---
 .../spring/kafka/MyCustomMessageConverter.java     |  2 +-
 .../de/example/spring/kafka/ReceiverConfig.java    |  4 ++--
 .../spring-cloud-stream-kafka-example-bom/pom.xml  | 25 +++++++++++++++++++++-
 3 files changed, 27 insertions(+), 4 deletions(-)

diff --git a/SpringJava/Kafka/SpringCloudStream/spring-cloud-stream-kafka-consumer-example/src/main/java/de/example/spring/kafka/MyCustomMessageConverter.java b/SpringJava/Kafka/SpringCloudStream/spring-cloud-stream-kafka-consumer-example/src/main/java/de/example/spring/kafka/MyCustomMessageConverter.java
index 51eae29..13667ad 100644
--- a/SpringJava/Kafka/SpringCloudStream/spring-cloud-stream-kafka-consumer-example/src/main/java/de/example/spring/kafka/MyCustomMessageConverter.java
+++ b/SpringJava/Kafka/SpringCloudStream/spring-cloud-stream-kafka-consumer-example/src/main/java/de/example/spring/kafka/MyCustomMessageConverter.java
@@ -21,6 +21,6 @@ public class MyCustomMessageConverter extends AbstractMessageConverter {
 
     logger.info("convertFromInternal, payload: " + payload);
 
-    return payload;
+	return (payload instanceof String ? payload : new String((byte[]) payload));
   }
 }
diff --git a/SpringJava/Kafka/SpringCloudStream/spring-cloud-stream-kafka-consumer-example/src/main/java/de/example/spring/kafka/ReceiverConfig.java b/SpringJava/Kafka/SpringCloudStream/spring-cloud-stream-kafka-consumer-example/src/main/java/de/example/spring/kafka/ReceiverConfig.java
index 5a9d1e2..3788a7b 100644
--- a/SpringJava/Kafka/SpringCloudStream/spring-cloud-stream-kafka-consumer-example/src/main/java/de/example/spring/kafka/ReceiverConfig.java
+++ b/SpringJava/Kafka/SpringCloudStream/spring-cloud-stream-kafka-consumer-example/src/main/java/de/example/spring/kafka/ReceiverConfig.java
@@ -3,7 +3,7 @@ package de.example.spring.kafka;
 import org.springframework.cloud.stream.schema.avro.AvroSchemaMessageConverter;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
-import org.springframework.core.io.ClassPathResource;
+//import org.springframework.core.io.ClassPathResource;
 import org.springframework.messaging.converter.MessageConverter;
 import org.springframework.util.MimeType;
 
@@ -28,5 +28,5 @@ public class ReceiverConfig {
       AvroSchemaMessageConverter converter = new AvroSchemaMessageConverter(MimeType.valueOf("avro/bytes"));
       //converter.setSchemaLocation(new ClassPathResource("schemas/User.avro"));
       return converter;
-    }
+  }
 }
diff --git a/SpringJava/Kafka/SpringCloudStream/spring-cloud-stream-kafka-example-bom/pom.xml b/SpringJava/Kafka/SpringCloudStream/spring-cloud-stream-kafka-example-bom/pom.xml
index ebbd9eb..769c420 100644
--- a/SpringJava/Kafka/SpringCloudStream/spring-cloud-stream-kafka-example-bom/pom.xml
+++ b/SpringJava/Kafka/SpringCloudStream/spring-cloud-stream-kafka-example-bom/pom.xml
@@ -39,7 +39,30 @@
             <artifactId>spring-cloud-starter-stream-kafka</artifactId>
             <version>1.1.1.RELEASE</version>
         </dependency>
-    
+
+        <!-- Required for using Kafka 0.10.1.1 -->
+        <dependency>
+            <groupId>org.springframework.kafka</groupId>
+            <artifactId>spring-kafka</artifactId>
+            <version>1.1.2.RELEASE</version>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.integration</groupId>
+            <artifactId>spring-integration-kafka</artifactId>
+            <version>2.1.0.RELEASE</version>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.kafka</groupId>
+            <artifactId>kafka_2.11</artifactId>
+            <version>0.10.1.1</version>
+            <exclusions>
+                <exclusion>
+                    <groupId>org.slf4j</groupId>
+                    <artifactId>slf4j-log4j12</artifactId>
+                </exclusion>
+            </exclusions>
+        </dependency>
+
         <!-- Required for using Apache Avro -->
         <dependency>
             <groupId>org.springframework.cloud</groupId>
-- 
2.1.4