Я пытаюсь установить потребителя Kafka, используя spring-kaka-2.2.0 и spring-boot-2.1.0, но у меня странная проблема, говорящая о том, что ConsumerFactory bean не найден, но все же он есть в конфигурации.
Iесть два типа сообщений в теме Student и Professor Я пытаюсь использовать их, используя @KafkaListener и @KafkaHandler, но не зашло так далеко
Вот мой конфиг с операторами импорта
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
@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);
}
}
Приложение не удалось запустить
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.
Но если внести незначительные изменения, удаление@Bean на consumerFactory() метод и непосредственное создание экземпляров работает, но все же он не берет свойства из consumerConfigs() Map.он пытается сопоставить значение по умолчанию localhost:9092, но я пытаюсь подключить удаленный кластер.
factory.setConsumerFactory(new DefaultKafkaConsumerFactory<>(consumerConfigs()));