Q: spring Kafka ConsumerFactory bean not found I'm trying to establish the Kafka consumer using spring-kaka-2.2.0 and spring-boot-2.1.0 but I'm having the weird issue saying ConsumerFactory bean not found but still it is there in config.
I have two types of messages in topic Student and Professor I'm trying to consumes them using @KafkaListener and @KafkaHandler but not gone so far
Here is my config with import statements
package com.kafka.configuration;
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.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.listener.ContainerProperties.AckMode;
import org.springframework.kafka.support.converter.BatchMessagingMessageConverter;
import org.springframework.kafka.support.converter.StringJsonMessageConverter;
import org.springframework.kafka.support.serializer.JsonDeserializer;

@Configuration
public class KafkaConsumerConfig {

@Value("${kafka.consumer.bootstrap-server}")
private String bootstrapServer;

@Bean("consumerConfigs")
public Map<String, Object> consumerConfigs() {
    Map<String, Object> props = new HashMap<>();
    props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServer);
    props.put(ConsumerConfig.GROUP_ID_CONFIG, "test-3");
    props.put(JsonDeserializer.TYPE_MAPPINGS,
            "professor:com.kafka.model.Professor, student:com.kafka.model.Student");
    props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
    props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, JsonDeserializer.class);
    props.put(ConsumerConfig.CLIENT_ID_CONFIG, "test-3");
    props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
    props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, "1");
    return props;
}

@Bean("consumerFactory")
public ConsumerFactory<String, Object> consumerFactory() {
    return new DefaultKafkaConsumerFactory<>(consumerConfigs());
}

@Bean("kafkaListenerContainerFactory")
@ConditionalOnMissingBean(name = "KafkaAutoConfiguration")
public ConcurrentKafkaListenerContainerFactory<String, Object> kafkaListenerContainerFactory() {
    ConcurrentKafkaListenerContainerFactory<String, Object> factory = new ConcurrentKafkaListenerContainerFactory<>();
    factory.setConsumerFactory(consumerFactory());
    factory.setBatchListener(true);
    factory.setMessageConverter(new BatchMessagingMessageConverter(converter()));
    factory.getContainerProperties().setAckMode(AckMode.BATCH);
    return factory;
}

@Bean
public StringJsonMessageConverter converter() {
    return new StringJsonMessageConverter();
   }

}

Service
 @Service
 @KafkaListener(topics = "test-events",id = "kafkaListenerContainerFactory")
 public class KafkaConsumerService {

@KafkaHandler
public void student(List<Student> stu) {
    System.out.println(stu);
}

@KafkaHandler
public void professor(List<Professor> pro) {
    System.out.println(pro);
}

@KafkaHandler
public void listenDefault(List<Object> obj) {
    System.out.println(obj);
    }

 }


APPLICATION FAILED TO START

Description:

Parameter 1 of method kafkaListenerContainerFactory in org.springframework.boot.autoconfigure.kafka.KafkaAnnotationDrivenConfiguration required a bean of type 'org.springframework.kafka.core.ConsumerFactory' that could not be found.

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


 Action:

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


But if make minor changes removing @Bean on consumerFactory() method and directly instating is working But still it is not taking properties from consumerConfigs() Map. it is trying to map default to localhost:9092 but I'm trying to connect remote cluster. 
 factory.setConsumerFactory(new DefaultKafkaConsumerFactory<>(consumerConfigs()));


A: If you want to override boot's default container factory, you must name it kafkaListenerContainerFactory since its condition is on the presence of a bean with that name.
@Bean
@ConditionalOnMissingBean(name = "kafkaListenerContainerFactory")
public ConcurrentKafkaListenerContainerFactory<?, ?> kafkaListenerContainerFactory(

That said, it's not clear why you are wiring up your own consumer and container factories since everything you are doing can be configured in application.yml (or .properties) and by adding a @Bean for the message converter (boot will wire it in for you).
Your ConsumerFactory does not have the correct generic types for the Boot autowired factory.
