Vfrdtyky

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.mobicule</groupId>

    <artifactId>spring-boot-kafka-consumer</artifactId>

    <version>0.0.1-SNAPSHOT</version>

    <packaging>jar</packaging>



    <name>spring-boot-kafka-consumer</name>

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



    <parent>

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

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

        <version>2.1.0.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</artifactId>

        </dependency>

        <dependency>

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

            <artifactId>spring-kafka</artifactId>

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

KafkaConfiguration.java

package com.mobicule.springbootkafkaconsumer.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;



@EnableKafka

@Configuration

public class KafkaConfiguration {



    @Bean

    public ConsumerFactory<String, String> consumerFactory()

    {

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



        config.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");

        config.put(ConsumerConfig.GROUP_ID_CONFIG, "group_id");

        config.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);

        config.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);





        return new DefaultKafkaConsumerFactory<String, String>(config);

    }



    @Bean

    public ConcurrentKafkaListenerContainerFactory<String, String> kafkaConsumerFactory()

    {

        ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<String, String>();



        factory.setConsumerFactory(consumerFactory());

        return factory;

    }







}

KafkaConsumer.java

package com.mobicule.springbootkafkaconsumer.listener;



import org.springframework.kafka.annotation.KafkaListener;

import org.springframework.stereotype.Component;



@Component

public class KafkaConsumer {



    @KafkaListener(topics="ConsumerDemo",groupId="group_id")

    public void consume(String message)

    {

        System.out.println("In kafka consumer "+message);

    }



}

I have configured Basic Kafka Producer project using Spring Initializer (without server)

1) I produced a message "Hello World" on topic "ConsumerConfig" using terminal

Then when i executed my consumer project i got error as

The following candidates were found but could not be injected:

    - Bean method 'kafkaConsumerFactory' in 'KafkaAutoConfiguration' not loaded because @ConditionalOnMissingBean (types: org.springframework.kafka.core.ConsumerFactory; SearchStrategy: all) found beans of type 'org.springframework.kafka.core.ConsumerFactory' consumerFactory

    - User-defined bean method 'consumerFactory' in 'KafkaConfiguration'





Action:



Consider revisiting the entries above or defining a bean of type 'org.springframework.kafka.core.ConsumerFactory' in your configuration.

So i searched and i got a link which said put
@SpringBootApplication(exclude = KafkaAutoConfiguration.class)

Even after putting the above in my code i got an error as

Description:

A component required a bean named 'kafkaListenerContainerFactory' that could not be found.





Action:



Consider defining a bean named 'kafkaListenerContainerFactory' in your configuration.

What am i missing here ? Do i need to do anymore configurations?

If you use Spring Boot, you don't need that custom ConsumerFactory, neither ConcurrentKafkaListenerContainerFactory. You should fully rely on the auto-configuration and appropriate configuration properties from the spring.kafka namespace: https://docs.spring.io/spring-boot/docs/2.1.0.RELEASE/reference/htmlsingle/#boot-features-kafka

If you still insist to exclude KafkaAutoConfiguration and do everything manually, you really need to name your ConcurrentKafkaListenerContainerFactory with the requested kafkaListenerContainerFactory bean name: https://docs.spring.io/spring-kafka/docs/2.2.0.RELEASE/reference/html/_reference.html#kafka-listener-annotation