Fix Error Handling

demo 2/pom.xml

@@ -4,11 +4,11 @@
 	<modelVersion>4.0.0</modelVersion>
 
 	<groupId>com.example</groupId>
-	<artifactId>demo</artifactId>
+	<artifactId>demoSCStErrors</artifactId>
 	<version>0.0.1-SNAPSHOT</version>
 	<packaging>jar</packaging>
 
-	<name>demo</name>
+	<name>demoSCStErrors</name>
 	<description>Demo project for Spring Boot</description>
 
 	<parent>
@@ -29,6 +29,25 @@
 		<dependency>
 			<groupId>org.springframework.cloud</groupId>
 			<artifactId>spring-cloud-stream-binder-kafka</artifactId>
+			<version>1.3.0.RELEASE</version>
+		</dependency>
+
+		<dependency>
+			<groupId>org.springframework.cloud</groupId>
+			<artifactId>spring-cloud-stream-binder-kafka-core</artifactId>
+			<version>1.3.0.RELEASE</version>
+		</dependency>
+
+		<dependency>
+			<groupId>org.springframework.cloud</groupId>
+			<artifactId>spring-cloud-stream</artifactId>
+			<version>1.3.0.RELEASE</version>
+		</dependency>
+
+		<dependency>
+			<groupId>org.springframework.cloud</groupId>
+			<artifactId>spring-cloud-stream-codec</artifactId>
+			<version>1.3.0.RELEASE</version>
 		</dependency>
 
 		<dependency>

demo 2/src/main/java/com/example/demo/DemoApplication.java

@@ -2,10 +2,8 @@
 
 import org.springframework.boot.SpringApplication;
 import org.springframework.boot.autoconfigure.SpringBootApplication;
-import org.springframework.integration.config.EnableIntegration;
 
 @SpringBootApplication
-@EnableIntegration
 public class DemoApplication {
 	public static void main(String[] args) {
 		SpringApplication.run(DemoApplication.class, args);

demo 2/src/main/java/com/example/demo/ErrorTransformerViaSCS.java

@@ -1,20 +1,14 @@
 package com.example.demo;
 
 
-import org.springframework.cloud.stream.annotation.EnableBinding;
-import org.springframework.cloud.stream.annotation.StreamListener;
-import org.springframework.cloud.stream.messaging.Processor;
 import org.springframework.messaging.Message;
-import org.springframework.messaging.handler.annotation.SendTo;
 
-import static com.example.demo.CustomDlqMessageChannel.CUSTOM_DLQ_TOPIC;
-
-@EnableBinding(Processor.class)
+//@EnableBinding(Processor.class)
 public class ErrorTransformerViaSCS {
     public static final String ERROR_CHANNEL = "errorChannel";
 
-    @StreamListener(ERROR_CHANNEL)
-    @SendTo(CUSTOM_DLQ_TOPIC)
+//    @StreamListener(ERROR_CHANNEL)
+//    @SendTo(CUSTOM_DLQ_TOPIC)
     public ErrorWrapper onError(Message message) {
         return new ErrorWrapper();
     }

demo 2/src/main/java/com/example/demo/ErrorTransformerViaSI.java

@@ -1,16 +1,20 @@
 package com.example.demo;
 
+import static com.example.demo.ErrorTransformerViaSCS.ERROR_CHANNEL;
+
+import org.springframework.cloud.stream.annotation.EnableBinding;
 import org.springframework.integration.annotation.ServiceActivator;
 import org.springframework.messaging.Message;
 import org.springframework.stereotype.Service;
 
-import static com.example.demo.ErrorTransformerViaSCS.ERROR_CHANNEL;
-
 @Service
+@EnableBinding(CustomDlqMessageChannel.class)
 public class ErrorTransformerViaSI {
 
     @ServiceActivator(inputChannel=ERROR_CHANNEL, outputChannel="customDlqTopic")
     public ErrorWrapper onError(Message message) {
         return new ErrorWrapper();
     }
+
 }
+

demo 2/src/main/resources/application.yml

@@ -3,11 +3,18 @@ spring:
     stream:
         bindings:
           input:
+            destination: testIn
             consumer:
-              autoCommitOnError: true
+              header-mode: raw #so we can send from the console consumer
           custom_dlq:
             destination: custom_dlq
         kafka:
-           binder:
-             brokers: 192.168.99.100
-             zkNodes: 192.168.99.100
+          bindings:
+            input:
+              autoCommitOnError: true
+#           binder:
+#             brokers: 192.168.99.100
+#             zkNodes: 192.168.99.100
+#logging:
+#  level:
+#    org.springframework.integration: debug