建设工程公开招标网站,西部数码成品网站,专业制作网站的基本步骤,网页设计于制作课程标准在测试的过程中#xff0c;心血来潮#xff0c;想要测试下新topic中还没被消费的消息。专门查了下ai#xff0c;奈何一本正经的胡说八道#xff0c;浪费了点时间。现在记录下#xff1a; 解决topic缺失时项目无法启动 #xff0c; 报错#xff1a; Topic(s) [……] is/a… 在测试的过程中心血来潮想要测试下新topic中还没被消费的消息。专门查了下ai奈何一本正经的胡说八道浪费了点时间。现在记录下 解决topic缺失时项目无法启动 报错 Topic(s) [……] is/are not present and missingTopicsFatal is true 指定消息消费的区间
Configuration
Primary
public class CommonKafkaConfig extends KafkaProperties {Value(${spring.kafka.concurrency})public int concurrency;Value(${spring.kafka.poll-timeout-ms})public long pollTimeout;Value(${spring.kafka.consumer.auto-offset-reset})private String autoOffsetReset;Value(${spring.kafka.consumer.auto-commit-interval-ms})private String autoCommitInterval;Value(${spring.kafka.consumer.bootstrap-servers})private String mpSyncBootstrapServers;Value(${spring.kafka.consumer.cloud-bootstrap-servers})private String cloudSyncBootstrapServers;Value(${spring.kafka.consumer.group-id})private String mpSyncGroupId;Value(${spring.kafka.consumer.max-pool-records})private String maxPoolRecords;Value(${spring.profiles.active})String env;Bean(name kafkaListenerContainerFactory)KafkaListenerContainerFactoryConcurrentMessageListenerContainerString, String kafkaListenerContainerFactory() {ConcurrentKafkaListenerContainerFactoryString, String factory new ConcurrentKafkaListenerContainerFactory();factory.setConsumerFactory(ConsumerFactory());factory.setConcurrency(concurrency);factory.getContainerProperties().setPollTimeout(pollTimeout);//忽略不存在的topic, 针对报警 // Topic(s) [……] is/are not present and missingTopicsFatal is truefactory.getContainerProperties().setMissingTopicsFatal(false);return factory;}private ConsumerFactoryString, String ConsumerFactory() {return new DefaultKafkaConsumerFactory(ConsumerConfigs());}private MapString, Object ConsumerConfigs() {MapString, Object props new HashMap();props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, getBootStrap());// 配置的值为latest 消费最新的消息。 // earliest: 表示当没有初始偏移量或者偏移量无效时消费者会从最早的可用消息开始消费props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, latest);props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, autoCommitInterval);props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);props.put(ConsumerConfig.GROUP_ID_CONFIG, mpSyncGroupId);props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, maxPoolRecords);return props;}public String getBootStrap() {if (env.contains(cloud)) return cloudSyncBootstrapServers;return mpSyncBootstrapServers;}private MapString, Object producerConfigs() {MapString, Object props new HashMap();props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, getBootStrap());props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);return props;}private ProducerFactoryString, String producerFactory() {return new DefaultKafkaProducerFactory(producerConfigs());}Bean(name commonKafkaTemplate)public KafkaTemplateString, String kafkaTemplate() {return new KafkaTemplate(producerFactory());}
}问题1日志里会有大量的[warn],如下
2024-04-02 16:58:32.906 WARN 15652 --- [ errorHandler-4-C-1] org.apache.kafka.clients.NetworkClient : [Consumer clientIdconsumer-69, groupIderrorHandler] Error while fetching metadata with correlation id 973 : {mp.publish.grab.high.priorityUNKNOWN_TOPIC_OR_PARTITION, mp.publish.otherUNKNOWN_TOPIC_OR_PARTITION, mp.publish.grab.low.priorityUNKNOWN_TOPIC_OR_PARTITION}
可以在logback.xml中加一行配置忽略日志 logger nameorg.apache.kafka.clients.NetworkClient levelERROR/
