Q1:Failed to start bean 'org.springframework.kafka.config.internalKafkaListenerEndpointRegistry'
A1:是因為沒有在注入時未對其進行配置導致,注冊參照如下
/**
* Created by wolf 2018/12/1
*/
@Configuration
@EnableKafka
public class kafkatemplateConfig {
@Value("${spring.kafka.bootstrap-servers}")
private String bootstrapServers;
@Bean
public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String,String>> kafkaListenerContainerFactory(){
ConcurrentKafkaListenerContainerFactory<String,String> factory = new ConcurrentKafkaListenerContainerFactory<>();
factory.setConsumerFactory(consumerFactory());
factory.setConcurrency(3);
factory.getContainerProperties().setPollTimeout(5000);
return factory;
}
@Bean
public ConsumerFactory<String,String> consumerFactory(){
return new DefaultKafkaConsumerFactory<>(consumerConfigs());
}
@Bean
public Map<String,Object> consumerConfigs(){
Map<String,Object> propsMap = new HashMap<>();
propsMap.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,this.bootstrapServers);
propsMap.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,false);//自行控制提交offset
propsMap.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG,"100");//提交延遲毫秒數
propsMap.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG,"15000");//執行超時時間
propsMap.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
propsMap.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
propsMap.put(ConsumerConfig.GROUP_ID_CONFIG,"dsafas");
propsMap.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,"latest");//開始消費位置
return propsMap;
}
@Bean
public Consumer listener(){
return new Consumer();
}
}
Q2:a:Could not instantiate class org.springframework.kafka.support.serializer.JsonDeserializer
b:org.springframework.kafka.support.serializer.JsonDeserializer with modifiers "protected"
c:If the serializationyou can also enable trust all (*)
A2:看下是因為要為監聽者得到的對象 在進行序列化與反序列化 要進行設置在上述配置類中如下兩列針對該問題
propsMap.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
propsMap.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
Q3:java.lang.IllegalArgumentException: Magic v1 does not support record headers
A3:版本問題 我其實遇到時是用的 springboot.version: 2.1.0.RELEASE 部署的kafka組件版本為較新版kafka_2.11-2.1.0 與舊版kafka_2.12-0.10.2.1 最終將springboot.version換成1.5.10.RELEASE
