mindol1004/spring-cloud-stream-kafka-communication
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Adding kafka communication using spring cloud streams kafka

Browse Source
This commit is contained in:
amrutprabhu
2021-12-05 16:48:17 +01:00
parent 59b8b44d68
commit 43cc49e87c
7 changed files with 181 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

65
spring-cloud-stream-kafka-communication/pom.xml Normal file
View File

@@ -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>

27
spring-cloud-stream-kafka-communication/src/main/java/com/amrut/prabhu/SpringCloudStreamKafkaApplication.java Normal file
View File

@@ -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");
}
}

4
spring-cloud-stream-kafka-communication/src/main/java/com/amrut/prabhu/dto/Message.java Executable file
View File

@@ -0,0 +1,4 @@
package com.amrut.prabhu.dto;
public record Message(String name) {
}

23
spring-cloud-stream-kafka-communication/src/main/java/com/amrut/prabhu/dto/coverters/MessageDeSerializer.java Normal file
View File

@@ -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);
}
}
}

22
spring-cloud-stream-kafka-communication/src/main/java/com/amrut/prabhu/dto/coverters/MessageSerializer.java Normal file
View File

@@ -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);
}
}
}

27
spring-cloud-stream-kafka-communication/src/main/resources/application.yml Normal file
View File

@@ -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

13
spring-cloud-stream-kafka-communication/src/test/java/com/amrut/prabhu/SpringCloudStreamKafkaApplicationTests.java Normal file
View File

@@ -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() {
}
}