From 43cc49e87cd856bdf035e7dc488706be61388e97 Mon Sep 17 00:00:00 2001
From: amrutprabhu <amrutprabhu42@gmail.com>
Date: Sun, 5 Dec 2021 16:48:17 +0100
Subject: [PATCH] Adding kafka communication using spring cloud streams kafka

---
 .../pom.xml                                   | 65 +++++++++++++++++++
 .../SpringCloudStreamKafkaApplication.java    | 27 ++++++++
 .../java/com/amrut/prabhu/dto/Message.java    |  4 ++
 .../dto/coverters/MessageDeSerializer.java    | 23 +++++++
 .../dto/coverters/MessageSerializer.java      | 22 +++++++
 .../src/main/resources/application.yml        | 27 ++++++++
 ...pringCloudStreamKafkaApplicationTests.java | 13 ++++
 7 files changed, 181 insertions(+)
 create mode 100644 spring-cloud-stream-kafka-communication/pom.xml
 create mode 100644 spring-cloud-stream-kafka-communication/src/main/java/com/amrut/prabhu/SpringCloudStreamKafkaApplication.java
 create mode 100755 spring-cloud-stream-kafka-communication/src/main/java/com/amrut/prabhu/dto/Message.java
 create mode 100644 spring-cloud-stream-kafka-communication/src/main/java/com/amrut/prabhu/dto/coverters/MessageDeSerializer.java
 create mode 100644 spring-cloud-stream-kafka-communication/src/main/java/com/amrut/prabhu/dto/coverters/MessageSerializer.java
 create mode 100644 spring-cloud-stream-kafka-communication/src/main/resources/application.yml
 create mode 100644 spring-cloud-stream-kafka-communication/src/test/java/com/amrut/prabhu/SpringCloudStreamKafkaApplicationTests.java

diff --git a/spring-cloud-stream-kafka-communication/pom.xml b/spring-cloud-stream-kafka-communication/pom.xml
new file mode 100644
index 0000000..874172f
--- /dev/null
+++ b/spring-cloud-stream-kafka-communication/pom.xml
@@ -0,0 +1,65 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
+	<modelVersion>4.0.0</modelVersion>
+	<parent>
+		<groupId>org.springframework.boot</groupId>
+		<artifactId>spring-boot-starter-parent</artifactId>
+		<version>2.6.1</version>
+<!--		<version>2.4.4</version>-->
+		<relativePath/> <!-- lookup parent from repository -->
+	</parent>
+	<groupId>com.amrut.prabhu</groupId>
+	<artifactId>spring-cloud-stream-kafka-communication</artifactId>
+	<version>0.0.1-SNAPSHOT</version>
+	<name>spring-cloud-stream-kafka-communication</name>
+	<description>Kafka communication with spring cloud Streams</description>
+	<properties>
+		<java.version>17</java.version>
+		<spring-cloud.version>2021.0.0</spring-cloud.version>
+<!--		<spring-cloud.version>2020.0.2</spring-cloud.version>-->
+	</properties>
+	<dependencies>
+		<dependency>
+			<groupId>org.apache.kafka</groupId>
+			<artifactId>kafka-streams</artifactId>
+		</dependency>
+		<dependency>
+			<groupId>org.springframework.cloud</groupId>
+			<artifactId>spring-cloud-stream-binder-kafka</artifactId>
+		</dependency>
+		<dependency>
+			<groupId>org.springframework.boot</groupId>
+			<artifactId>spring-boot-starter-test</artifactId>
+			<scope>test</scope>
+		</dependency>
+		<dependency>
+			<groupId>org.springframework.cloud</groupId>
+			<artifactId>spring-cloud-stream</artifactId>
+			<scope>test</scope>
+			<classifier>test-binder</classifier>
+			<type>test-jar</type>
+		</dependency>
+	</dependencies>
+	<dependencyManagement>
+		<dependencies>
+			<dependency>
+				<groupId>org.springframework.cloud</groupId>
+				<artifactId>spring-cloud-dependencies</artifactId>
+				<version>${spring-cloud.version}</version>
+				<type>pom</type>
+				<scope>import</scope>
+			</dependency>
+		</dependencies>
+	</dependencyManagement>
+
+	<build>
+		<plugins>
+			<plugin>
+				<groupId>org.springframework.boot</groupId>
+				<artifactId>spring-boot-maven-plugin</artifactId>
+			</plugin>
+		</plugins>
+	</build>
+
+</project>
diff --git a/spring-cloud-stream-kafka-communication/src/main/java/com/amrut/prabhu/SpringCloudStreamKafkaApplication.java b/spring-cloud-stream-kafka-communication/src/main/java/com/amrut/prabhu/SpringCloudStreamKafkaApplication.java
new file mode 100644
index 0000000..c65abd6
--- /dev/null
+++ b/spring-cloud-stream-kafka-communication/src/main/java/com/amrut/prabhu/SpringCloudStreamKafkaApplication.java
@@ -0,0 +1,27 @@
+package com.amrut.prabhu;
+
+import com.amrut.prabhu.dto.Message;
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.context.annotation.Bean;
+
+import java.util.function.Consumer;
+import java.util.function.Supplier;
+
+@SpringBootApplication
+public class SpringCloudStreamKafkaApplication {
+
+    public static void main(String[] args) {
+        SpringApplication.run(SpringCloudStreamKafkaApplication.class, args);
+    }
+
+    @Bean
+    public Consumer<Message> consumer() {
+        return message -> System.out.println("received " + message);
+    }
+
+    @Bean
+    public Supplier<Message> producer() {
+        return () -> new Message(" jack from Streams");
+    }
+}
diff --git a/spring-cloud-stream-kafka-communication/src/main/java/com/amrut/prabhu/dto/Message.java b/spring-cloud-stream-kafka-communication/src/main/java/com/amrut/prabhu/dto/Message.java
new file mode 100755
index 0000000..5ce2b81
--- /dev/null
+++ b/spring-cloud-stream-kafka-communication/src/main/java/com/amrut/prabhu/dto/Message.java
@@ -0,0 +1,4 @@
+package com.amrut.prabhu.dto;
+
+public record Message(String name) {
+}
diff --git a/spring-cloud-stream-kafka-communication/src/main/java/com/amrut/prabhu/dto/coverters/MessageDeSerializer.java b/spring-cloud-stream-kafka-communication/src/main/java/com/amrut/prabhu/dto/coverters/MessageDeSerializer.java
new file mode 100644
index 0000000..157ef31
--- /dev/null
+++ b/spring-cloud-stream-kafka-communication/src/main/java/com/amrut/prabhu/dto/coverters/MessageDeSerializer.java
@@ -0,0 +1,23 @@
+package com.amrut.prabhu.dto.coverters;
+
+import com.amrut.prabhu.dto.Message;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.kafka.common.errors.SerializationException;
+import org.apache.kafka.common.serialization.Deserializer;
+
+import java.io.IOException;
+
+public class MessageDeSerializer implements Deserializer<Message> {
+
+    private final ObjectMapper objectMapper = new ObjectMapper();
+
+    @Override
+    public Message deserialize(String topic, byte[] data) {
+        try {
+
+            return objectMapper.readValue(new String(data), Message.class);
+        } catch (IOException e) {
+            throw new SerializationException(e);
+        }
+    }
+}
diff --git a/spring-cloud-stream-kafka-communication/src/main/java/com/amrut/prabhu/dto/coverters/MessageSerializer.java b/spring-cloud-stream-kafka-communication/src/main/java/com/amrut/prabhu/dto/coverters/MessageSerializer.java
new file mode 100644
index 0000000..382548c
--- /dev/null
+++ b/spring-cloud-stream-kafka-communication/src/main/java/com/amrut/prabhu/dto/coverters/MessageSerializer.java
@@ -0,0 +1,22 @@
+package com.amrut.prabhu.dto.coverters;
+
+import com.amrut.prabhu.dto.Message;
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.kafka.common.errors.SerializationException;
+import org.apache.kafka.common.serialization.Serializer;
+
+public class MessageSerializer implements Serializer<Message> {
+
+    private final ObjectMapper objectMapper = new ObjectMapper();
+
+    @Override
+    public byte[] serialize(String topic, Message data) {
+        try {
+            return objectMapper.writeValueAsBytes(data);
+        } catch (JsonProcessingException e) {
+            throw new SerializationException(e);
+        }
+
+    }
+}
diff --git a/spring-cloud-stream-kafka-communication/src/main/resources/application.yml b/spring-cloud-stream-kafka-communication/src/main/resources/application.yml
new file mode 100644
index 0000000..561edb8
--- /dev/null
+++ b/spring-cloud-stream-kafka-communication/src/main/resources/application.yml
@@ -0,0 +1,27 @@
+spring:
+  cloud:
+    function:
+      definition: consumer;producer
+    stream:
+      kafka:
+        bindings:
+          producer-out-0:
+            producer:
+              configuration:
+                value.serializer: com.amrut.prabhu.dto.coverters.MessageSerializer
+          consumer-in-0:
+            consumer:
+              configuration:
+                value.deserializer: com.amrut.prabhu.dto.coverters.MessageDeSerializer
+        binder:
+          brokers: localhost:9092
+            
+      bindings:
+        producer-out-0:
+            destination : first-topic
+            producer:
+                useNativeEncoding: true # This enables using the custom serializer
+        consumer-in-0:
+            destination : first-topic
+            consumer:
+              use-native-decoding: true # This enables using the custom deserializer
diff --git a/spring-cloud-stream-kafka-communication/src/test/java/com/amrut/prabhu/SpringCloudStreamKafkaApplicationTests.java b/spring-cloud-stream-kafka-communication/src/test/java/com/amrut/prabhu/SpringCloudStreamKafkaApplicationTests.java
new file mode 100644
index 0000000..5ae18c4
--- /dev/null
+++ b/spring-cloud-stream-kafka-communication/src/test/java/com/amrut/prabhu/SpringCloudStreamKafkaApplicationTests.java
@@ -0,0 +1,13 @@
+package com.amrut.prabhu;
+
+import org.junit.jupiter.api.Test;
+import org.springframework.boot.test.context.SpringBootTest;
+
+@SpringBootTest
+class SpringCloudStreamKafkaApplicationTests {
+
+	@Test
+	void contextLoads() {
+	}
+
+}