kafka-producer and consumer application added

kafka-consumer-master/README.md

+# kafka-consumer
+Apache Kafka Consumer Example using SpringBoot

kafka-consumer-master/pom.xml

+<?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 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+	<modelVersion>4.0.0</modelVersion>
+
+	<groupId>com.example</groupId>
+	<artifactId>kafka-consumer</artifactId>
+	<version>0.0.1-SNAPSHOT</version>
+	<packaging>jar</packaging>
+
+	<name>kafka-consumer</name>
+	<description>listen message from topic </description>
+
+	<parent>
+		<groupId>org.springframework.boot</groupId>
+		<artifactId>spring-boot-starter-parent</artifactId>
+		<version>2.0.2.RELEASE</version>
+		<relativePath /> <!-- lookup parent from repository -->
+	</parent>
+
+	<properties>
+		<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
+		<project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
+		<java.version>1.8</java.version>
+	</properties>
+
+	<dependencies>
+		<dependency>
+			<groupId>org.springframework.boot</groupId>
+			<artifactId>spring-boot-starter-web</artifactId>
+		</dependency>
+
+		<dependency>
+			<groupId>org.springframework.kafka</groupId>
+			<artifactId>spring-kafka</artifactId>
+			<version>2.1.4.RELEASE</version>
+		</dependency>
+		
+		<dependency>
+            <groupId>org.springframework.kafka</groupId>
+            <artifactId>spring-kafka-test</artifactId>
+            <version>2.1.4.RELEASE</version>
+            <scope>test</scope>
+        </dependency>
+
+		<dependency>
+			<groupId>org.springframework.boot</groupId>
+			<artifactId>spring-boot-devtools</artifactId>
+			<scope>runtime</scope>
+		</dependency>
+		<dependency>
+			<groupId>org.projectlombok</groupId>
+			<artifactId>lombok</artifactId>
+			<optional>true</optional>
+		</dependency>
+		<dependency>
+			<groupId>org.springframework.boot</groupId>
+			<artifactId>spring-boot-starter-test</artifactId>
+			<scope>test</scope>
+		</dependency>
+	</dependencies>
+
+	<build>
+		<plugins>
+			<plugin>
+				<groupId>org.springframework.boot</groupId>
+				<artifactId>spring-boot-maven-plugin</artifactId>
+			</plugin>
+		</plugins>
+	</build>
+
+
+</project>

kafka-consumer-master/src/main/java/com/nisum/spring/kafka/api/KafkaConsumerApplication.java

+package com.nisum.spring.kafka.api;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.kafka.annotation.KafkaListener;
+import org.springframework.kafka.annotation.PartitionOffset;
+import org.springframework.kafka.annotation.TopicPartition;
+import org.springframework.kafka.support.KafkaHeaders;
+import org.springframework.messaging.handler.annotation.Header;
+import org.springframework.messaging.handler.annotation.Payload;
+import org.springframework.web.bind.annotation.GetMapping;
+import org.springframework.web.bind.annotation.RestController;
+
+@SpringBootApplication
+@RestController
+public class KafkaConsumerApplication {
+
+	private static final Logger LOGGER = LoggerFactory.getLogger(KafkaConsumerApplication.class);
+	List<String> messages = new ArrayList<>();
+
+	User userFromTopic = null;
+
+	@GetMapping("/consumeStringMessage")
+	public List<String> consumeMsg() {
+		return messages;
+	}
+
+	/*
+	 * @GetMapping("/consumeJsonMessage") public User consumeJsonMessage() { return
+	 * userFromTopic; }
+	 */
+
+	@KafkaListener(topics = "sample1", containerFactory = "kafkaListenerContainerFactory")
+	public List<String> getMsgFromTopic(String data) {
+		LOGGER.info(String.format("Event message received -> %s", data));
+		messages.add(data);
+		return messages;
+	}
+
+	// https://reflectoring.io/spring-boot-kafka/
+	// https://hevodata.com/learn/spring-kafka-consumer/
+	// https://memorynotfound.com/spring-kafka-adding-custom-header-kafka-message-example/
+	@KafkaListener(topicPartitions = @TopicPartition(topic = "test",
+			partitionOffsets = {
+			@PartitionOffset(partition = "0", initialOffset = "0"),
+			@PartitionOffset(partition = "1", initialOffset = "0")
+			}),
+			containerFactory = "kafkaListenerContainerFactory")
+	public void listenToPartition(@Payload String message, @Header(KafkaHeaders.RECEIVED_PARTITION_ID) int partition,
+			@Header(KafkaHeaders.OFFSET) int offset) { 
+
+		System.out.println("Received Message: " + message + " from partition: " + partition + " offset : " + offset);
+	}
+
+	/*
+	 * @KafkaListener(groupId = "nisum-2", topics = "sample1", containerFactory =
+	 * "userKafkaListenerContainerFactory") public User getJsonMsgFromTopic(User
+	 * user) { userFromTopic = user; return userFromTopic; }
+	 */
+
+	public static void main(String[] args) {
+		SpringApplication.run(KafkaConsumerApplication.class, args);
+	}
+}

kafka-consumer-master/src/main/java/com/nisum/spring/kafka/api/User.java

+package com.nisum.spring.kafka.api;
+
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+@Data
+@AllArgsConstructor
+@NoArgsConstructor
+public class User {
+	private int id;
+	private String name;
+	private String[] address;
+
+}

kafka-consumer-master/src/main/java/com/nisum/spring/kafka/api/config/KafkaConsumerConfig.java

+package com.nisum.spring.kafka.api.config;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.common.serialization.StringDeserializer;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.kafka.annotation.EnableKafka;
+import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
+import org.springframework.kafka.core.ConsumerFactory;
+import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
+import org.springframework.kafka.support.serializer.JsonDeserializer;
+
+import com.nisum.spring.kafka.api.User;
+
+@Configuration
+@EnableKafka
+public class KafkaConsumerConfig {
+
+	// config for String plain text
+
+	@Bean
+	public ConsumerFactory<String, String> consumerFactory() {
+		Map<String, Object> configs = new HashMap<>();
+		configs.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
+		configs.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
+		configs.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
+		configs.put(ConsumerConfig.GROUP_ID_CONFIG, "nisum-1");
+		return new DefaultKafkaConsumerFactory<>(configs);
+	}
+
+	@Bean
+	public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory() {
+		ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<String, String>();
+		factory.setConsumerFactory(consumerFactory());
+		return factory;
+	}
+
+	// config for json data
+
+	@Bean
+	public ConsumerFactory<String, User> userConsumerFactory() {
+		Map<String, Object> configs = new HashMap<>();
+		configs.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
+		configs.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
+		configs.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, JsonDeserializer.class);
+		configs.put(ConsumerConfig.GROUP_ID_CONFIG, "nisum-2");
+		return new DefaultKafkaConsumerFactory<>(configs, new StringDeserializer(), new JsonDeserializer<>(User.class));
+	}
+
+	@Bean
+	public ConcurrentKafkaListenerContainerFactory<String, User> userKafkaListenerContainerFactory() {
+		ConcurrentKafkaListenerContainerFactory<String, User> factory = new ConcurrentKafkaListenerContainerFactory<String, User>();
+		factory.setConsumerFactory(userConsumerFactory());
+		return factory;
+	}
+
+}

kafka-consumer-master/src/main/resources/application.properties

+server.port=8080
\ No newline at end of file

kafka-consumer-master/src/test/java/com/nisum/spring/kafka/api/KafkaConsumerApplicationTests.java

+package com.nisum.spring.kafka.api;
+
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.test.context.junit4.SpringRunner;
+
+@RunWith(SpringRunner.class)
+@SpringBootTest
+public class KafkaConsumerApplicationTests {
+
+	@Test
+	public void contextLoads() {
+	}
+
+}

kafka-publisher-master/pom.xml

+<?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 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+	<modelVersion>4.0.0</modelVersion>
+
+	<groupId>com.example</groupId>
+	<artifactId>kafka-publisher</artifactId>
+	<version>0.0.1-SNAPSHOT</version>
+	<packaging>jar</packaging>
+
+	<name>kafka-publisher</name>
+	<description>Publish message using kafka</description>
+
+	<parent>
+		<groupId>org.springframework.boot</groupId>
+		<artifactId>spring-boot-starter-parent</artifactId>
+		<version>2.0.2.RELEASE</version>
+		<relativePath/> <!-- lookup parent from repository -->
+	</parent>
+
+	<properties>
+		<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
+		<project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
+		<java.version>1.8</java.version>
+	</properties>
+
+	<dependencies>
+		<dependency>
+			<groupId>org.springframework.boot</groupId>
+			<artifactId>spring-boot-starter-web</artifactId>
+		</dependency>
+		<dependency>
+			<groupId>org.springframework.kafka</groupId>
+			<artifactId>spring-kafka</artifactId>
+		</dependency>
+
+		<dependency>
+			<groupId>org.springframework.boot</groupId>
+			<artifactId>spring-boot-devtools</artifactId>
+			<scope>runtime</scope>
+		</dependency>
+		<dependency>
+			<groupId>org.projectlombok</groupId>
+			<artifactId>lombok</artifactId>
+			<optional>true</optional>
+		</dependency>
+		<dependency>
+			<groupId>org.springframework.boot</groupId>
+			<artifactId>spring-boot-starter-test</artifactId>
+			<scope>test</scope>
+		</dependency>
+	</dependencies>
+
+	<build>
+		<plugins>
+			<plugin>
+				<groupId>org.springframework.boot</groupId>
+				<artifactId>spring-boot-maven-plugin</artifactId>
+			</plugin>
+		</plugins>
+	</build>
+
+
+</project>

kafka-publisher-master/src/main/java/com/nisum/spring/kafka/api/KafkaPublisherApplication.java

+package com.nisum.spring.kafka.api;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import org.apache.kafka.clients.producer.KafkaProducer;
+import org.apache.kafka.clients.producer.Producer;
+import org.apache.kafka.clients.producer.ProducerConfig;
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.common.serialization.StringSerializer;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.web.bind.annotation.GetMapping;
+import org.springframework.web.bind.annotation.PathVariable;
+import org.springframework.web.bind.annotation.RestController;
+
+import com.nisum.spring.kafka.api.config.CustomPartitioner;
+
+@SpringBootApplication
+@RestController
+public class KafkaPublisherApplication {
+
+	@Autowired
+	private KafkaTemplate<String, Object> template;
+	
+	@Autowired
+	private Producer<String, String> producer;
+
+	private String topic = "sample1";
+
+	@GetMapping("/publish/{name}")
+	public String publishMessage(@PathVariable String name) {
+		template.send(topic, "Hi " + name + " Welcome to Nisum");
+		return "Data published";
+	}
+	 
+	@GetMapping("/publishInPartision/{name}")
+	public String publishMessageInSpecificPartision(@PathVariable String name) {
+		
+		producer.send(new ProducerRecord<>("test",0, "hai how are you","value"+1 ));
+		//producer.close();
+		
+		return "Data published";
+	}
+
+	/*
+	 * @GetMapping("/publishJson") public String publishMessage() { User user = new
+	 * User(2532, "User88", new String[] { "Bangalore", "BTM", "house 90" });
+	 * template.send(topic, user); return "Json Data published"; }
+	 */
+	public static void main(String[] args) {
+		SpringApplication.run(KafkaPublisherApplication.class, args);
+	}
+}

kafka-publisher-master/src/main/java/com/nisum/spring/kafka/api/User.java

+package com.nisum.spring.kafka.api;
+
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+@Data
+@AllArgsConstructor
+@NoArgsConstructor
+public class User {
+	private int id;
+	private String name;
+	private String[] address;
+
+}

kafka-publisher-master/src/main/java/com/nisum/spring/kafka/api/config/CustomPartitioner.java

+package com.nisum.spring.kafka.api.config;
+
+import java.util.Objects;
+
+import org.apache.kafka.clients.producer.internals.DefaultPartitioner;
+import org.apache.kafka.common.Cluster;
+
+public class CustomPartitioner extends DefaultPartitioner{
+	
+	@Override
+	  public int partition(String topic, Object key, byte[] keyBytes,
+	  Object value, byte[] valueBytes,
+	      Cluster cluster) {
+
+	    String partitionKey = "";
+	    if (Objects.nonNull(key)) {
+	      String bookingKey = (String) key;
+	      //Ignore bookingKey.getBookingDate()
+	      keyBytes = partitionKey.getBytes();
+	    }
+	    return super.partition(topic, partitionKey, keyBytes, value,
+	    valueBytes, cluster);
+	  }
+
+}

kafka-publisher-master/src/main/java/com/nisum/spring/kafka/api/config/KafkaPublisherConfig.java

+package com.nisum.spring.kafka.api.config;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import org.apache.kafka.clients.producer.KafkaProducer;
+import org.apache.kafka.clients.producer.Producer;
+import org.apache.kafka.clients.producer.ProducerConfig;
+import org.apache.kafka.common.serialization.StringSerializer;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.kafka.core.DefaultKafkaProducerFactory;
+import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.kafka.core.ProducerFactory;
+import org.springframework.kafka.support.serializer.JsonSerializer;
+
+@Configuration
+public class KafkaPublisherConfig {
+
+	@Bean
+	public ProducerFactory<String, Object> producerFactory() {
+		Map<String, Object> configs = new HashMap<>();
+		configs.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
+		configs.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
+		configs.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
+		return new DefaultKafkaProducerFactory<String, Object>(configs);
+	}
+
+	@Bean
+	public KafkaTemplate<String, Object> kafkaTemplate() {
+		return new KafkaTemplate<>(producerFactory());
+	}
+
+	@Bean
+	public Producer<String, String> kafkaProducerFactory() {
+		Map<String, Object> configs = new HashMap<>();
+		configs.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
+		configs.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
+		configs.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
+		//https://www.clairvoyant.ai/blog/writing-custom-partitioner-for-apache-kafka
+		//configs.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, CustomPartitioner.class);
+		return new KafkaProducer<String, String>(configs);
+	}
+
+}

kafka-publisher-master/src/main/resources/application.properties

+server.port=9091
\ No newline at end of file

kafka-publisher-master/src/test/java/com/nisum/spring/kafka/api/KafkaPublisherApplicationTests.java

+package com.nisum.spring.kafka.api;
+
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.test.context.junit4.SpringRunner;
+
+@RunWith(SpringRunner.class)
+@SpringBootTest
+public class KafkaPublisherApplicationTests {
+
+	@Test
+	public void contextLoads() {
+	}
+
+}