Sunday, November 17, 2024
Google search engine
HomeLanguagesJavaSpring Boot – Consume Message Through Kafka, Save into ElasticSearch, and Plot...
LanguagesJava

Spring Boot – Consume Message Through Kafka, Save into ElasticSearch, and Plot into Grafana

Nicole Veronica
By Nicole Veronica
0
0

In this article, we are going to make a program to produce some data using Kafka Producer which will consume by the Kafka Consumer and save into the elastic search DB, and later on, plot that JSON data into the Grafana dashboard. Start with configuring all the required software and tool.

Requirements

  • Java 8
  • Spring Boot 2.7.0
  • Kafka 3.2.0
  • Elastic search DB 7.17.3

Note: Why do we mention the version?

Version compatibility is of paramount importance in the case of Elastic and Spring Boot. If your elastic-search version doesn’t match with the spring boot version or vice versa, then you face problems in configuring both. Below is the list of version compatibility : 

Spring Data Elasticsearch

Elastic-Search

Spring framework

Spring Boot

4.4.X

7.17.3

5.3.X

2.7.X

4.3.X

7.15.2

5.3.X

2.6.X

4.2.X

7.12.0

5.3.X

2.5.X

4.1.X

7.9.3

5.3.2

2.4.X

4.0.X

7.6.2

5.2.12

2.3.X

3.2.X

6.8.12

5.2.12

2.2.X

3.1.X

6.2.2

5.1.19

2.1.X

3.0.X

5.5.0

5.0.13

2.0.X

2.1.X

2.4.0

4.3.25

1.5.X

Download links: 

Download and extract from your system.

Create Kafka Producer Application

Create a spring boot project named Producer.

ProducerApplication.class

Java




package com.example.demo;
 
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
 
@SpringBootApplication
public class ProducerApplication {
    public static void main(String[] args) {
        SpringApplication.run(ProducerApplication.class, args);
    }
   
}
 



















KafkaProducerConfig.java


In this class, we have provided the configuration of Kafka. 






Java

















































package com.example.demo.config;


 


import java.util.HashMap;


import java.util.Map;


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;


import com.example.demo.model.User;


@Configuration


public class KafkaProducerConfig {


   


    @Bean


    public ProducerFactory<String, User> userProducerFactory() {


        Map<String, Object> configProps = new HashMap<>();


        configProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"localhost:9092");


        configProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,


                        StringSerializer.class.getName());


        configProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,


                        JsonSerializer.class.getName());


        return new DefaultKafkaProducerFactory<>(configProps);


    }


   


    @Bean


    public KafkaTemplate<String, User> userKafkaTemplate() {


        return new KafkaTemplate<>(userProducerFactory());


    }


   


}








































User.java


Model class where we store the user information.






Java

















































package com.example.demo.model;


 


public class User {


    int id;


    String name;


    String pdate;


    public User() {


        super();


    }


    public User(int id, String name, String pdate) {


        super();


        this.id = id;


        this.name = name;


        this.pdate = pdate;


    }


    public int getId() {


        return id;


    }


    public void setId(int id) {


        this.id = id;


    }


    public String getName() {


        return name;


    }


    public void setName(String name) {


        this.name = name;


    }


    public String getPdate() {


        return pdate;


    }


    public void setPdate(String pdate) {


        this.pdate = pdate;


    }


}








































KafkaService.java


Service class uses kafka template to send the data to the consumer. 








Java

















































package com.example.demo.service;


 


import java.util.List;


import org.slf4j.Logger;


import org.slf4j.LoggerFactory;


import org.springframework.beans.factory.annotation.Autowired;


import org.springframework.kafka.core.KafkaTemplate;


import org.springframework.stereotype.Service;


import com.example.demo.model.User;


 


@Service


public class KafkaService {


    private final Logger LOG = LoggerFactory.getLogger(KafkaService.class);


    @Autowired


    private KafkaTemplate<String, User> kafkaTemplate;


    String kafkaTopic = "gfg";


    public void send(User user) {


        LOG.info("Sending User Json Serializer : {}", user);


        kafkaTemplate.send(kafkaTopic, user);


    }


    public void sendList(List<User> userList) {


        LOG.info("Sending UserList Json Serializer : {}", userList);


        for (User user : userList) {


            kafkaTemplate.send(kafkaTopic, user);


        }


    }


}








































ProducerController.java






Java

















































package com.example.demo.controller;


 


import java.util.List;


import org.springframework.beans.factory.annotation.Autowired;


import org.springframework.web.bind.annotation.PostMapping;


import org.springframework.web.bind.annotation.RequestBody;


import org.springframework.web.bind.annotation.RestController;


 


import com.example.demo.model.User;


import com.example.demo.service.KafkaService;


 


@RestController


public class ProducerController {


    @Autowired


    KafkaService kafkaProducer;


   


    @PostMapping("/producer")


    public String sendMessage(@RequestBody User user) {


        kafkaProducer.send(user);


        return "Message sent successfully to the Kafka topic shubham";


    }


   


    @PostMapping("/producerlist")


    public String sendMessage(@RequestBody List<User> user) {


        kafkaProducer.sendList(user);


        return "Message sent successfully to the Kafka topic shubham";


    }


}








































pom.xml






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


                        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.7.0</version>


        <relativePath/>


    </parent>


    <groupId>com.example</groupId>


    <artifactId>Producer</artifactId>


    <version>0.0.1-SNAPSHOT</version>


    <name>Producer</name>


    <description>Demo project for Spring Boot</description>


    <properties>


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


            <optional>true</optional>


        </dependency>


        <dependency>


            <groupId>org.springframework.boot</groupId>


            <artifactId>spring-boot-starter-test</artifactId>


            <scope>test</scope>


        </dependency>


        <dependency>


            <groupId>org.springframework.kafka</groupId>


            <artifactId>spring-kafka-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>








































application.properties


server.port=1234


Create Kafka Consumer and Configure the ElastisSearch Application


Create another spring boot application named ElasticConsumer.


ComsumerApplication.java






Java

















































package com.example.demo;


 


import org.springframework.beans.factory.annotation.Autowired;


import org.springframework.boot.SpringApplication;


import org.springframework.boot.autoconfigure.SpringBootApplication;


import org.springframework.kafka.annotation.KafkaListener;


import org.springframework.web.bind.annotation.GetMapping;


import org.springframework.web.bind.annotation.RestController;


import com.example.demo.model.User;


import com.example.demo.service.KafkaUserService;


 


@SpringBootApplication


@RestController


public class ConsumerApplication {


    @Autowired


    KafkaUserService kafkaUserService;


   


    public static void main(String[] args) {


        SpringApplication.run(ConsumerApplication.class, args);


    }


   


    @KafkaListener(topics = "gfg", groupId = "gfg-group")


    public void listen(User user) {


        System.out.println("Received User information : " + user.toString());


        try {


            kafkaUserService.saveUser(user);


        } catch (Exception e) {


            e.printStackTrace();   


        }


    }


   


    @GetMapping("/getElasticUserFromKafka")


    public Iterable<User> findAllUser() {


        return kafkaUserService.findAllUsers();


    }


}








































KafkaConsumerConfig.java






Java

















































package com.example.demo.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.example.demo.model.User;


 


@EnableKafka


@Configuration


public class kafkaConsumerConfig {


    @Bean


    public ConsumerFactory<String, User> consumerFactory() {


        Map<String, Object> props = new HashMap<>();


        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:9092");


        props.put(ConsumerConfig.GROUP_ID_CONFIG, "gfg-group");


        return new DefaultKafkaConsumerFactory<>(props,


            new StringDeserializer(), new JsonDeserializer<>(User.class));


    }


   


    @Bean


    public ConcurrentKafkaListenerContainerFactory<String, User>


      kafkaListenerContainerFactory() {


        ConcurrentKafkaListenerContainerFactory<String, User> factory =


          new ConcurrentKafkaListenerContainerFactory<>();


        factory.setConsumerFactory(consumerFactory());


        return factory;


    }


   


}








































User.java




@Document – specifies our index.
@Id – represents the field _id of our document and it is unique for each message.
@Field – represents a different type of field that might be in our data.








Java

















































package com.example.demo.model;


 


import java.util.Date;


import org.springframework.data.annotation.Id;


import org.springframework.data.elasticsearch.annotations.Document;


import org.springframework.data.elasticsearch.annotations.Field;


import org.springframework.data.elasticsearch.annotations.FieldType;


import com.google.gson.Gson;


 


@Document(indexName = "kafkauser")


public class User {


    @Id


    int id;


    @Field(type = FieldType.Text, name = "name")


    String name;


    @Field(type = FieldType.Date, name = "pdate")


    Date pdate;


    public User() {


        super();


    }


    public User(int id, String name, Date pdate) {


        super();


        this.id = id;


        this.name = name;


        this.pdate = pdate;


    }


    public Date getPdate() {


        return pdate;


    }


    public void setPdate(Date pdate) {


        this.pdate = pdate;


    }


    public int getId() {


        return id;


    }


    public void setId(int id) {


        this.id = id;


    }


    public String getName() {


        return name;


    }


    public void setName(String name) {


        this.name = name;


    }


    @Override


    public String toString() {


        return new Gson().toJson(this);


    }


}








































KafkaUserRepository.java








Java

















































package com.example.demo.repository;


 


import org.springframework.data.elasticsearch.repository.ElasticsearchRepository;


import org.springframework.stereotype.Repository;


import com.example.demo.model.User;


 


@Repository


public interface KafkaUserRepository extends ElasticsearchRepository<User,String> {


   


}








































KafkaUserService.java






Java

















































package com.example.demo.service;


 


import org.springframework.beans.factory.annotation.Autowired;


import org.springframework.stereotype.Service;


import com.example.demo.model.User;


import com.example.demo.repository.KafkaUserRepository;


 


@Service


public class KafkaUserService {


    @Autowired


    private KafkaUserRepository edao;


   


    public void saveUser(User user) {


        edao.save(user);


    }


   


    public Iterable<User> findAllUsers() {


    return edao.findAll();


    }   


}








































How to Run?




Elastis-Search:


    

  • Run elastissearch.bat using cmd -> E:\elasticsearch-7.17.3-windows-x86_64\elasticsearch-7.17.3\bin> elasticsearch.bat
    • 

  • open browser and type –  http://localhost:9200
    • 



The output look like this:


{
“name”: “DESKTOP-S6UTE8M”,
“cluster_name”: “elasticsearch”,
“cluster_uuid”: “VDlwyl2WQhCX7_lLwWm9Kg”,
“version”: {
“number”: “7.17.3”,
“build_flavor”: “default”,
“build_type”: “zip”,
“build_hash”: “5ad023604c8d7416c9eb6c0eadb62b14e766caff”,
“build_date”: “2022-04-19T08:11:19.070913226Z”,
“build_snapshot”: false,
“lucene_version”: “8.11.1”,
“minimum_wire_compatibility_version”: “6.8.0”,
“minimum_index_compatibility_version”: “6.0.0-beta1”
},
“tagline”: “You Know, for Search”
}


Kafka: 


In Kafka first we have to run zookeeper and after that kafkaserver. In windows we have to run .bat file and in the case of linux we have to run .sh file 


    

  • Open cmd 
    • 

  • Navigate under kafka folder
    • 

  • E:\kafka_2.12-3.2.0\kafka_2.12-3.2.0> .\bin\windows\zookeeper-server-start.bat .\config\zookeeper.properties
    • 

  • E:\kafka_2.12-3.2.0\kafka_2.12-3.2.0> .\bin\windows\kafka-server-start.bat .\config\server.properties
    • 





pom.xml






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


                        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.7.0</version>


        <relativePath />


    </parent>


    <groupId>com.example</groupId>


    <artifactId>Consumer</artifactId>


    <version>0.0.1-SNAPSHOT</version>


    <name>Consumer</name>


    <description>Demo project for Spring Boot</description>


    <properties>


        <java.version>1.8</java.version>


    </properties>


    <dependencies>


        <dependency>


            <groupId>org.springframework.boot</groupId>


            <artifactId>spring-boot-starter-web</artifactId>


        </dependency>


        <dependency>


            <groupId>com.google.code.gson</groupId>


            <artifactId>gson</artifactId>


 


        </dependency>


        <dependency>


            <groupId>org.springframework.kafka</groupId>


            <artifactId>spring-kafka</artifactId>


        </dependency>


         


        <dependency>


            <groupId>org.springframework.boot</groupId>


            <artifactId>spring-boot-starter-data-elasticsearch</artifactId>


        </dependency>


 


        <dependency>


            <groupId>org.springframework.boot</groupId>


            <artifactId>spring-boot-devtools</artifactId>


            <scope>runtime</scope>


            <optional>true</optional>


        </dependency>


        <dependency>


            <groupId>org.springframework.boot</groupId>


            <artifactId>spring-boot-starter-test</artifactId>


            <scope>test</scope>


        </dependency>


        <dependency>


            <groupId>org.springframework.kafka</groupId>


            <artifactId>spring-kafka-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>








































Run Producer and ElasticConsumer Spring Application


Send JSON data using postman. Here data is sent by the Producer app to ElasticConsumer. And in the ElasticConsumer console data would be printed and saved into ElasticSearchDB in the form of JSON.


Producer App APIs:


    

  • Send single object -> http://localhost:1234/producer
    • 

  • Send list of objects -> http://localhost:1234/producerlist
    • 





 




ElasticConsumer app APIs


    

  • Fetch all records from elastic db -> localhost:8080/getElasticUserFromKafka
    • 



Grafana Dashboard


Grafana dashboard is running on http://localhots:3000. Watch the configuration video below.




Output


Output Video 1:




Output Video 2:




Some ElasticSearch APIs


    

  • To show the records of index -> http://localhost:9200/<index_name>/_search
    • 

  • To Delete index -> http://localhost:9200/<index_name>
    • 

  • List all indices -> http://localhost:9200/_cat/indices
    • 

  • Show schema of index -> http://localhost:9200/<index_name>
    • 
















Previous article
Java.io.StringWriter class in Java
Next article
Spring – Using SQL Scripts with Spring JDBC + JPA + HSQLDB



Nicole Veronica
Nicole Veronica


RELATED ARTICLES











Most Popular
Load more


Recent Comments


	

강서구출장마사지 on How to store XML data into a MySQL database using Python?
	
 

	

금천구출장마사지 on How to store XML data into a MySQL database using Python?
	
 

	

nightwish.southeast.cz on Google says it won’t keep your Pixel during a repair if you’re caught using non-OEM parts
	
 

	

광명출장안마 on How to store XML data into a MySQL database using Python?
	
 

	

광명출장안마 on How to store XML data into a MySQL database using Python?
	
 

	

출장오피 on How to store XML data into a MySQL database using Python?
	
 

	

부천출장안마 on How to store XML data into a MySQL database using Python?
	
 

	

구월동출장안마 on How to store XML data into a MySQL database using Python?
	
 

	

강서구출장안마 on How to store XML data into a MySQL database using Python?
	
 

	

헬로출장 on How to store XML data into a MySQL database using Python?
	
 

	

오산출장안마 on How to store XML data into a MySQL database using Python?
	
 

	

광명출장마사지 on How to store XML data into a MySQL database using Python?
	
 

	

마포출장 on How to store XML data into a MySQL database using Python?
	
 

	

안양출장마사지 on How to store XML data into a MySQL database using Python?
	
 

	

gKTdhA on 5 reasons why I won’t switch away from Google Photos
	
 

	

부천출장안마 on How to store XML data into a MySQL database using Python?
	
 

	

동탄출장안마 on How to store XML data into a MySQL database using Python?
	
 

	

0a1Mq7 on Wander: An add-on for Apple’s Shortcuts app to install the Odyssey jailbreak
	
 

	

서울출장안마 on How to store XML data into a MySQL database using Python?
	
 

	

분당출장안마 on How to store XML data into a MySQL database using Python?
	
 

	

부천출장안마 on How to store XML data into a MySQL database using Python?
	
 

	

출장 오피 on How to store XML data into a MySQL database using Python?
	
 

	

화곡동출장마사지 on How to store XML data into a MySQL database using Python?
	
 

	

Gilda on Wander: An add-on for Apple’s Shortcuts app to install the Odyssey jailbreak
	
 

	

강서구출장마사지 on How to store XML data into a MySQL database using Python?
	
 

	

고양출장안마 on How to store XML data into a MySQL database using Python?
	
 

	

화성출장마사지 on How to store XML data into a MySQL database using Python?
	
 

	

천호동출장마사지 on How to store XML data into a MySQL database using Python?
	
 

	

June P. D. Alvarez on RedSn0w Updated to Fix iBooks DRM Issues
	
 

	

Litha on How to Install Siri on iPad 2
	
 
 
                                    

            

        

        
    
 


    

                

            














EDITOR PICKS





POPULAR POSTS





POPULAR CATEGORY










 







ABOUT US



Newspaper is your news, entertainment, music fashion website. We provide you with the latest breaking news and videos straight from the entertainment industry.



Contact us: hello@geeksforgeeks.org






FOLLOW US











© Lazyr0ar 2022