From f886be669fdd50d6fe9643a817d23fc900307a4c Mon Sep 17 00:00:00 2001
From: Gustavo Martin Morcuende <gu.martinm@gmail.com>
Date: Sun, 29 Jan 2017 19:55:59 +0100
Subject: [PATCH] Required default constructor for Jackson deserialization.

---
 .../java/de/example/spring/kafka/MyCustomMessageConverter.java    | 2 --
 .../src/main/java/de/example/spring/kafka/Product.java            | 7 +++++++
 .../src/main/resources/application.yml                            | 3 +++
 .../src/main/resources/application.yml                            | 8 ++++----
 4 files changed, 14 insertions(+), 6 deletions(-)

diff --git a/SpringJava/Kafka/SpringCloudStream/spring-cloud-stream-kafka-common-example/src/main/java/de/example/spring/kafka/MyCustomMessageConverter.java b/SpringJava/Kafka/SpringCloudStream/spring-cloud-stream-kafka-common-example/src/main/java/de/example/spring/kafka/MyCustomMessageConverter.java
index f489302..419ae06 100644
--- a/SpringJava/Kafka/SpringCloudStream/spring-cloud-stream-kafka-common-example/src/main/java/de/example/spring/kafka/MyCustomMessageConverter.java
+++ b/SpringJava/Kafka/SpringCloudStream/spring-cloud-stream-kafka-common-example/src/main/java/de/example/spring/kafka/MyCustomMessageConverter.java
@@ -1,7 +1,5 @@
 package de.example.spring.kafka;
 
-import java.nio.charset.Charset;
-
 import org.springframework.messaging.converter.MappingJackson2MessageConverter;
 import org.springframework.util.MimeType;
 
diff --git a/SpringJava/Kafka/SpringCloudStream/spring-cloud-stream-kafka-common-example/src/main/java/de/example/spring/kafka/Product.java b/SpringJava/Kafka/SpringCloudStream/spring-cloud-stream-kafka-common-example/src/main/java/de/example/spring/kafka/Product.java
index 98564a5..6053200 100644
--- a/SpringJava/Kafka/SpringCloudStream/spring-cloud-stream-kafka-common-example/src/main/java/de/example/spring/kafka/Product.java
+++ b/SpringJava/Kafka/SpringCloudStream/spring-cloud-stream-kafka-common-example/src/main/java/de/example/spring/kafka/Product.java
@@ -4,6 +4,13 @@ public class Product {
 	private String name;
 	private String description;
 	
+	/**
+	 * Required for deserialization from Jackson.
+	 */
+	protected Product() {
+
+	}
+
 	public Product(String name, String description) {
 		this.name = name;
 		this.description = description;
diff --git a/SpringJava/Kafka/SpringCloudStream/spring-cloud-stream-kafka-consumer-example/src/main/resources/application.yml b/SpringJava/Kafka/SpringCloudStream/spring-cloud-stream-kafka-consumer-example/src/main/resources/application.yml
index 72eafba..f4f8991 100644
--- a/SpringJava/Kafka/SpringCloudStream/spring-cloud-stream-kafka-consumer-example/src/main/resources/application.yml
+++ b/SpringJava/Kafka/SpringCloudStream/spring-cloud-stream-kafka-consumer-example/src/main/resources/application.yml
@@ -11,6 +11,8 @@ kafka:
 spring:
   main:
     banner-mode: "off"
+    # We do not need Tomcat running (this application runs from console)
+    web-environment: false
   cloud:
     stream:
       bindings:
@@ -19,6 +21,7 @@ spring:
           destination: test
           group: helloworld
           contentType: application/json
+          startOffset: latest
           consumer:
             concurrency: 1
             partitioned: false
diff --git a/SpringJava/Kafka/SpringCloudStream/spring-cloud-stream-kafka-producer-example/src/main/resources/application.yml b/SpringJava/Kafka/SpringCloudStream/spring-cloud-stream-kafka-producer-example/src/main/resources/application.yml
index 07b1ebf..eff109c 100644
--- a/SpringJava/Kafka/SpringCloudStream/spring-cloud-stream-kafka-producer-example/src/main/resources/application.yml
+++ b/SpringJava/Kafka/SpringCloudStream/spring-cloud-stream-kafka-producer-example/src/main/resources/application.yml
@@ -7,7 +7,7 @@ logging:
 spring:
   main:
     banner-mode: "off"
-    # We do not need Tomcat running (this application runs as a client)
+    # We do not need Tomcat running (this application runs from console)
     web-environment: false
   cloud:
     stream:
@@ -15,10 +15,10 @@ spring:
         output:
           binder: kafka
           destination: test
+          # Requires work around, see: MyCustomMessageConverter
           contentType: application/json
-          # headerMode value is used only when input bindings :(
-          # headerMode: raw
-          #contentType: "application/example"
+          # Consumer (input) and producer (output) are Spring Cloud Stream applications :)
+          headerMode: embeddedHeaders
       instanceCount: 1
       instanceIndex: 0
       kafka:
-- 
2.1.4