为现有消费者服务Spring Boot应用程序创建生产者配置时出现问题

Question

我有一个成功消耗来自Kafka主题的消息的Spring引导服务。作为进一步开发的一部分，我需要在之前创建并运行良好的相同消费者服务中将消息发布到另一个Kafka主题。但是当我介绍Kafka生产者配置时，它会给我消费者配置中的错误，说

 WARN  | [main] | org.springframework.context.support.AbstractApplicationContext.refresh | Exception encountered during context initialization - cancelling refresh attempt: org.springframework.beans.factory.BeanCreationException: Error creating bean with name 'kafkaConsumerService' defined in file 
 [C:\Workspaces\springboot\test-app\target\classes\com\testapp\execution\carrier\km\consumer\KafkaConsumerService.class]: Initialization of bean failed; nested exception is org.springframework.beans.factory.UnsatisfiedDependencyException: Error creating bean with name 'kafkaListenerContainerFactory' defined in class path resource [com/testapp/execution/carrier/km/config/KafkaConsumerConfig.class]: 
Unsatisfied dependency expressed through method 'kafkaListenerContainerFactory' parameter 0; nested exception is org.springframework.beans.factory.NoSuchBeanDefinitionException: No qualifying bean of type 'org.springframework.kafka.core.KafkaTemplate<java.lang.String, com.testapp.execution.carrier.km.model.CarrierStopEBO>' 
available: expected at least 1 bean which qualifies as autowire candidate. Dependency annotations: {}


Description:

Parameter 0 of method kafkaListenerContainerFactory in com.testapp.execution.carrier.km.config.KafkaConsumerConfig required a bean of type 'org.springframework.kafka.core.KafkaTemplate' that could not be found.

The following candidates were found but could not be injected:
    - Bean method 'kafkaTemplate' in 'KafkaAutoConfiguration' not loaded because @ConditionalOnMissingBean (types: org.springframework.kafka.core.KafkaTemplate; SearchStrategy: all) found beans of type 'org.springframework.kafka.core.KafkaTemplate' kafkaTemplatetest


Action:

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

我的工作Kafka消费者配置，这是突然抛出问题后，介绍生产者配置

Kafka消费者配置。JAVA

@Slf4j
@EnableKafka
@Configuration
public class KafkaConsumerConfig {
    @Value("${spring.kafka.offset-reset-policy}")
    private String offsetResetPolicy;
    @Value("${spring.kafka.group-id}")
    private String groupId;
    @Value("${spring.kafka.bootstrap-servers}")
    private String bootstrapServers;

    @Value("${spring.kafka.max-poll-interval}")
    private Integer maxPollInterval;

    @Value("${spring.kafka.max-poll-records}")
    private Integer maxPollRecords;

    @Value("${spring.kafka.session-timeout}")
    private Integer sessionTimeout;
    @Value("${spring.kafka.trusted-packages}")
    private String trustedPacakges;

    public KafkaConsumerConfig() {
    }

    @Bean
    public Map<String, Object> consumerConfigs() {

        Map<String, Object> props = new HashMap<>();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, JsonDeserializer.class);
        props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, offsetResetPolicy);
        props.put(JsonDeserializer.TRUSTED_PACKAGES, trustedPacakges);
        props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, maxPollRecords);
        props.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, maxPollInterval);
        props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, sessionTimeout);//
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
        props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, SASL_PLAINTEXT);

        return props;
    }

    @Bean
    public ConsumerFactory<String, CarrierStopEBO> consumerFactory() {
        return new DefaultKafkaConsumerFactory<>(consumerConfigs(), new StringDeserializer(),
                new ErrorHandlingDeserializer2<>(new JsonDeserializer<>(carrierStopEBO.class, false)));
    }

    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, CarrierStopEBO> kafkaListenerContainerFactory(
            KafkaTemplate<String, CarrierStopEBO> kafkaTemplate1) {
        ConcurrentKafkaListenerContainerFactory<String, CarrierStopEBO> factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory());
        factory.getContainerProperties().setAckOnError(false);
        factory.getContainerProperties().setSyncCommits(true);
        factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);
        factory.setAckDiscarded(true);
        factory.getContainerProperties().setAuthorizationExceptionRetryInterval(Duration.ofMillis(30000));
  

        return factory;
    }


}

将其添加到服务后导致问题的生产者配置

Kafka普罗德里格。JAVA

@Slf4j
@EnableKafka
@Configuration
public class KafkaProducerConfig {

    @Value("${sni.kafka.bootstrap-servers}")
    private String bootstrapServers;

    @Value("${sni.kafka.producer.properties.security.protocol}")
    private String securityProtocol;

    @Value("${sni.kafka.producer.properties.sasl.mechanism}")
    private String saslMechanism;

    @Value("${sni.kafka.producer.properties.sasl.jaas.config}")
    private String jaasConfig;

    public KafkaProducerConfig() {
   }
    @Bean
    public ProducerFactory<String, CarrierModelDTO> producerFactory() {
        Map<String, Object> configProperties = new HashMap<String, Object>();
        configProperties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        configProperties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        configProperties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);
        configProperties.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, DefaultPartitioner.class);
        configProperties.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, securityProtocol);
        return new DefaultKafkaProducerFactory<String, CarrierModelDTO>(configProperties);
    }

    @Bean
    public KafkaTemplate<String, CarrierModelDTO> kafkaTemplatetest() {
        return new KafkaTemplate<String, CarrierModelDTO>(producerFactory());
    }
}

我能够找出消费者和生产者配置与下面的生产者配置代码之间存在一些冲突

 @Bean
    public KafkaTemplate<String, CarrierModelDTO> kafkaTemplatetest() {
        return new KafkaTemplate<String, CarrierModelDTO>(producerFactory());
    }

有人能就这个问题提出建议吗

Answer 1

你期待一个KafkaTemplate