spring 引导中的 PropertySource 似乎不起作用

Question

我无法使用 PropertySource 加载 StringDeserializer 和 StringSerializer，但会加载所有其他 spring.kafka 相关的东西。

这是 spring PropertySource 中的错误吗？

我的常用应用程序如下所示：

@Configuration
@PropertySource("classpath:/config/kafka.properties")
public class CommonKafkaAutoConfiguration {
}

以下配置在文件名 kafka.properties 时不会加载，但在 application.properties

时可以

spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer
spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer

有趣的是在两种情况下都加载了：

spring.kafka.consumer.max-poll-records=20

更新我可以看到它被 kafkaBinderDefaultProperties 覆盖

Answer 1

这对我来说很好...

@SpringBootApplication
@PropertySource("classpath:/config/kafka.properties")
public class So65311053Application {

    public static void main(String[] args) {
        SpringApplication.run(So65311053Application.class, args);
    }

    @Bean
    public ApplicationRunner runner(KafkaTemplate<String, String> template, ConsumerFactory<?, ?> cf) {
        return args -> {
            template.send("so65311053", "foo");

            System.out.println(template.getProducerFactory()
                    .getConfigurationProperties()
                    .get(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG));

            System.out.println(cf.getConfigurationProperties()
                    .get(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG));
        };
    }

    @Bean
    public NewTopic topic() {
        return TopicBuilder.name("so65311053").partitions(1).replicas(1).build();
    }

    @KafkaListener(id = "so65311053", topics = "so65311053")
    void listen(String in) {
        System.out.println(in);
    }

}

spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.IntegerDeserializer
spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.IntegerSerializer

spring.kafka.consumer.auto-offset-reset=earliest

2020-12-15 13:38:49.154  INFO 4612 --- [           main] o.a.k.clients.consumer.ConsumerConfig    : ConsumerConfig values: 
    allow.auto.create.topics = true
    auto.commit.interval.ms = 5000
    auto.offset.reset = earliest
    ...
    key.deserializer = class org.apache.kafka.common.serialization.IntegerDeserializer
    ...
2020-12-15 13:38:49.211  INFO 4612 --- [           main] o.a.k.clients.producer.ProducerConfig    : ProducerConfig values: 
    ...
    key.serializer = class org.apache.kafka.common.serialization.IntegerSerializer

class org.apache.kafka.common.serialization.IntegerSerializer
class org.apache.kafka.common.serialization.IntegerDeserializer
2020-12-15 13:38:49.349  INFO 4612 --- [o65311053-0-C-1] o.s.k.l.KafkaMessageListenerContainer    : so65311053: partitions assigned: [so65311053-0]
foo

Answer 2

错误是因为KafkaBinderEnvironmentPostProcessor

我通过创建自己的 EnvironmentPostProcessor 解决了这个问题

spring 引导中的 PropertySource 似乎不起作用

PropertySource in spring boot doesn't seem to work

spring

spring-boot

spring-cloud-stream

spring-kafka

spring-cloud-stream-binder-kafka