Using Kafka 0.10.1.1

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 (file)
--- 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 (file)
--- 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 (file)
--- 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>