。

专门查了下ai#xff0c;奈何一本正经的胡说八道#xff0c;浪费了点时间。

现在记录下#xff1a;

解决topic缺失时项目无法启动

在测试的过程中心血来潮想要测试下新topic中还没被消费的消息。

专门查了下ai奈何一本正经的胡说八道浪费了点时间。

现在记录下

解决topic缺失时项目无法启动

{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

ConcurrentKafkaListenerContainerFactory();factory.setConsumerFactory(ConsumerFactory());factory.setConcurrency(concurrency);factory.getContainerProperties().setPollTimeout(pollTimeout);//忽略不存在的topic,

针对报警

truefactory.getContainerProperties().setMissingTopicsFatal(false);return

factory;}private

DefaultKafkaConsumerFactory(ConsumerConfigs());}private

MapString,

HashMap();props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,

getBootStrap());//

表示当没有初始偏移量或者偏移量无效时消费者会从最早的可用消息开始消费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,

props;}public

cloudSyncBootstrapServers;return

mpSyncBootstrapServers;}private

MapString,

HashMap();props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,

getBootStrap());props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,

StringSerializer.class);props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,

props;}private

DefaultKafkaProducerFactory(producerConfigs());}Bean(name

KafkaTemplateString,

KafkaTemplate(producerFactory());}

2024-04-02

org.apache.kafka.clients.NetworkClient

[Consumer

{mp.publish.grab.high.priorityUNKNOWN_TOPIC_OR_PARTITION,

mp.publish.o***rUNKNOWN_TOPIC_OR_PARTITION,

mp.publish.grab.low.priorityUNKNOWN_TOPIC_OR_PARTITION}

logger

nameorg.apache.kafka.clients.NetworkClient

levelERROR/