BAEL-2885 Spring Cloud Stream with Kakfa (#7114)

* BAEL-2927 added logging configuration for tomcat logs

* BAEL-2885 coding done

* BAEL-2885 formating done

* BAEL-2885 scheam namespace fixed

* fixed logback issue and added group name in config

* updated code to use the correct header for partition key

* removed the partition count

* updated the code for given comments

* BAEL-2885 removed the schema registry client dependency

spring-cloud/spring-cloud-stream/spring-cloud-stream-kafka/pom.xml

@@ -0,0 +1,109 @@
+<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 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+
+
+    <artifactId>spring-cloud-stream-kafka</artifactId>
+    <name>spring-cloud-stream-kafka</name>
+    <description>Simple Spring Cloud Stream</description>
+    <packaging>jar</packaging>
+
+    <parent>
+        <groupId>org.springframework.boot</groupId>
+        <artifactId>spring-boot-starter-parent</artifactId>
+        <version>2.1.5.RELEASE</version>
+    </parent>
+
+    <properties>
+        <java.version>1.8</java.version>
+        <spring-cloud.version>Greenwich.SR1</spring-cloud.version>
+    </properties>
+
+    <dependencies>
+
+        <dependency>
+            <groupId>org.springframework.cloud</groupId>
+            <artifactId>spring-cloud-stream-binder-kafka</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>org.springframework.cloud</groupId>
+            <artifactId>spring-cloud-stream-schema</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>org.springframework.cloud</groupId>
+            <artifactId>spring-cloud-stream-test-support</artifactId>
+            <scope>test</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>io.confluent</groupId>
+            <artifactId>kafka-avro-serializer</artifactId>
+            <version>4.0.0</version>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.avro</groupId>
+            <artifactId>avro-compiler</artifactId>
+            <version>1.8.2</version>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.avro</groupId>
+            <artifactId>avro-maven-plugin</artifactId>
+            <version>1.8.2</version>
+        </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>
+            <plugin>
+                <groupId>org.apache.avro</groupId>
+                <artifactId>avro-maven-plugin</artifactId>
+                <version>1.8.2</version>
+                <executions>
+                    <execution>
+                        <id>schemas</id>
+                        <phase>generate-sources</phase>
+                        <goals>
+                            <goal>schema</goal>
+                            <goal>protocol</goal>
+                            <goal>idl-protocol</goal>
+                        </goals>
+                        <configuration>
+                            <sourceDirectory>${project.basedir}/src/main/resources/</sourceDirectory>
+                            <outputDirectory>${project.basedir}/src/main/java/</outputDirectory>
+                        </configuration>
+                    </execution>
+                </executions>
+            </plugin>
+        </plugins>
+    </build>
+
+    <repositories>
+        <repository>
+            <id>confluent</id>
+            <url>https://packages.confluent.io/maven/</url>
+        </repository>
+    </repositories>
+
+</project>

spring-cloud/spring-cloud-stream/spring-cloud-stream-kafka/src/main/java/com/baeldung/AvroKafkaApplication.java

@@ -0,0 +1,18 @@
+package com.baeldung;
+
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.cloud.stream.annotation.EnableBinding;
+import org.springframework.cloud.stream.messaging.Processor;
+import org.springframework.cloud.stream.schema.client.EnableSchemaRegistryClient;
+
+@SpringBootApplication
+@EnableBinding(Processor.class)
+@EnableSchemaRegistryClient
+public class AvroKafkaApplication {
+
+    public static void main(String[] args) {
+        SpringApplication.run(AvroKafkaApplication.class, args);
+    }
+
+}

spring-cloud/spring-cloud-stream/spring-cloud-stream-kafka/src/main/java/com/baeldung/config/SchemRegistryConfig.java

@@ -0,0 +1,18 @@
+package com.baeldung.config;
+
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.cloud.stream.schema.client.ConfluentSchemaRegistryClient;
+import org.springframework.cloud.stream.schema.client.SchemaRegistryClient;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+@Configuration
+public class SchemRegistryConfig {
+
+    @Bean
+    public SchemaRegistryClient schemaRegistryClient(@Value("${spring.cloud.stream.kafka.binder.producer-properties.schema.registry.url}") String endPoint) {
+        ConfluentSchemaRegistryClient client = new ConfluentSchemaRegistryClient();
+        client.setEndpoint(endPoint);
+        return client;
+    }
+}

spring-cloud/spring-cloud-stream/spring-cloud-stream-kafka/src/main/java/com/baeldung/consumer/AvroConsumer.java

@@ -0,0 +1,21 @@
+package com.baeldung.consumer;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.cloud.stream.annotation.StreamListener;
+import org.springframework.cloud.stream.messaging.Processor;
+import org.springframework.stereotype.Service;
+
+import com.baeldung.schema.Employee;
+
+@Service
+public class AvroConsumer {
+
+    private static final Logger LOGGER = LoggerFactory.getLogger(AvroConsumer.class);
+
+    @StreamListener(Processor.INPUT)
+    public void consumeEmployeeDetails(Employee employeeDetails) {
+        LOGGER.info("Let's process employee details: {}", employeeDetails);
+    }
+
+}

spring-cloud/spring-cloud-stream/spring-cloud-stream-kafka/src/main/java/com/baeldung/controller/AvroController.java

@@ -0,0 +1,22 @@
+package com.baeldung.controller;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.web.bind.annotation.PathVariable;
+import org.springframework.web.bind.annotation.PostMapping;
+import org.springframework.web.bind.annotation.RestController;
+
+import com.baeldung.producer.AvroProducer;
+
+@RestController
+public class AvroController {
+
+    @Autowired
+    private AvroProducer avroProducer;
+
+    @PostMapping("/employees/{id}/{firstName}/{lastName}")
+    public String producerAvroMessage(@PathVariable int id, @PathVariable String firstName, @PathVariable String lastName) {
+        avroProducer.produceEmployeeDetails(id, firstName, lastName);
+        return "Sent employee details to consumer";
+    }
+
+}

spring-cloud/spring-cloud-stream/spring-cloud-stream-kafka/src/main/java/com/baeldung/producer/AvroProducer.java

@@ -0,0 +1,42 @@
+package com.baeldung.producer;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.cloud.stream.messaging.Processor;
+import org.springframework.kafka.support.KafkaHeaders;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.support.MessageBuilder;
+import org.springframework.stereotype.Service;
+
+import com.baeldung.schema.Employee;
+import com.baeldung.schema.EmployeeKey;
+
+@Service
+public class AvroProducer {
+
+    @Autowired
+    private Processor processor;
+
+    public void produceEmployeeDetails(int empId, String firstName, String lastName) {
+
+        // creating employee details
+        Employee employee = new Employee();
+        employee.setId(empId);
+        employee.setFirstName(firstName);
+        employee.setLastName(lastName);
+        employee.setDepartment("IT");
+        employee.setDesignation("Engineer");
+
+        // creating partition key for kafka topic
+        EmployeeKey employeeKey = new EmployeeKey();
+        employeeKey.setId(empId);
+        employeeKey.setDepartmentName("IT");
+
+        Message<Employee> message = MessageBuilder.withPayload(employee)
+            .setHeader(KafkaHeaders.MESSAGE_KEY, employeeKey)
+            .build();
+
+        processor.output()
+            .send(message);
+    }
+
+}

spring-cloud/spring-cloud-stream/spring-cloud-stream-kafka/src/main/resources/application.yaml

@@ -0,0 +1,29 @@
+spring:
+  cloud:
+    stream:
+      default: 
+         producer: 
+          useNativeEncoding: true
+         consumer:  
+          useNativeEncoding: true     
+      bindings:
+        input:
+          destination: employee-details
+          content-type: application/*+avro
+          group: group-1
+          concurrency: 3
+        output:
+          destination: employee-details
+          content-type: application/*+avro
+      kafka:
+          binder:        
+           producer-properties:
+             key.serializer: io.confluent.kafka.serializers.KafkaAvroSerializer
+             value.serializer: io.confluent.kafka.serializers.KafkaAvroSerializer
+             schema.registry.url: http://localhost:8081 
+           consumer-properties:
+             key.deserializer: io.confluent.kafka.serializers.KafkaAvroDeserializer
+             value.deserializer: io.confluent.kafka.serializers.KafkaAvroDeserializer
+             schema.registry.url: http://localhost:8081
+             specific.avro.reader: true
+                           

spring-cloud/spring-cloud-stream/spring-cloud-stream-kafka/src/main/resources/employee-key-schema.avsc

@@ -0,0 +1,14 @@
+{
+	"type": "record",
+	"name": "EmployeeKey",
+	"namespace": "com.baeldung.schema",
+	"fields": [
+	 {
+		"name": "id",
+		"type": "int"
+	},
+	{
+		"name": "departmentName",
+		"type": "string"
+	}]
+}

spring-cloud/spring-cloud-stream/spring-cloud-stream-kafka/src/main/resources/employee-schema.avsc

@@ -0,0 +1,29 @@
+{
+	"type": "record",
+	"name": "Employee",
+	"namespace": "com.baeldung.schema",
+	"fields": [
+	{
+		"name": "id",
+		"type": "int"
+	},
+	{
+		"name": "firstName",
+		"type": "string"
+	},
+	{
+		"name": "lastName",
+		"type": "string"
+	},
+	{
+		"name": "department",
+		"type": "string",
+		 "default": "IT "
+	},
+	{
+		"name": "designation",
+		"type": "string",
+		 "default": "Software Engineer"
+	}
+	]
+}