Spring 从 application.yaml 启动 Kafka StreamsConfig 或 ConsumerConfig 不适用

Question

我有一个非常简单的 spring 引导项目和一个 KTable，我想在 application.yml 中自定义我的配置，但配置似乎没有应用。这是我的配置文件application.yml

spring:
  kafka:
    bootstrap-servers: ${KAFKA_BOOTSTRAP_SERVERS:localhost:9092}
    streams:
      application-id: ${APPLICATION_ID:train-builder-processor}
      buffered-records-per-partition: 50

    consumer:
      auto-offset-reset: earliest
      max-poll-records: ${MAX_POLL_RECORDS:50}
      max-poll-interval-ms: ${KAFKA_CONSUMER_MAX_POLL_INTERVAL_MS:1000}
      properties:
        spring:
          json:
            trusted:
              packages:
                - com.example.kafkastream

但是，当启动应用程序时，日志输出如下：

2022-03-03 08:20:06.992  INFO 32989 --- [           main] s.r.s.m.t.TrainBuilderApplication        : Starting TrainBuilderApplication using Java 16.0.2 on MAPFVFG90ZQQ05P with PID 32989 (/Users/xxx/dev/train-builder-processor/target/classes started by xxx in /Users/xxx/dev/train-builder-processor)
2022-03-03 08:20:06.995 DEBUG 32989 --- [           main] s.r.s.m.t.TrainBuilderApplication        : Running with Spring Boot v2.6.3, Spring v5.3.15
2022-03-03 08:20:06.995  INFO 32989 --- [           main] s.r.s.m.t.TrainBuilderApplication        : No active profile set, falling back to default profiles: default
2022-03-03 08:20:08.856  INFO 32989 --- [           main] org.apache.kafka.streams.StreamsConfig   : StreamsConfig values: 
    acceptable.recovery.lag = 10000
    application.id = test.train-builder-processor
    application.server = 
    bootstrap.servers = [localhost:9092]
    buffered.records.per.partition = 1000
... (a bunch of other configs)

消费者配置：

...
    max.poll.interval.ms = 300000
    max.poll.records = 1000
...

下面是简单的应用程序 class 我正在使用：

@EnableKafka
@EnableKafkaStreams
@SpringBootApplication
public class TrainBuilderApplication {

    ...

    @Autowired
    private TrainIdMapper trainIdMapper;

    @Autowired
    private TrainBuilder trainBuilder;

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

    @Bean
    public KTable<String, Train> trainTable(StreamsBuilder kStreamBuilder) {
        return kStreamBuilder
                .stream(Pattern.compile(sourceTopicsPattern), Consumed.with(Serdes.String(), myJsonSerde))
                .map(trainIdMapper)
                .filter((key, value) -> key != null)
                .groupByKey(Grouped.with(Serdes.String(), mySerde))
                .aggregate(() -> null, trainBuilder, trainStore);
    }
}

我的 application.yml 中的值似乎被忽略了。这可能是什么原因造成的？我错过了什么？提前致谢！

Answer 1

所以我在的帮助下弄明白了。

显然，consumer 和 producer 配置在使用 KStream 时与 streams 配置完全分开。要为 Kafka 流的 消费者设置特定属性 必须使用“附加属性”，如下所示：

spring:
  kafka:
    bootstrap-servers: ${KAFKA_BOOTSTRAP_SERVERS,localhost:9092}
    streams:
      application-id: ${APPLICATION_ID:train-builder-processor}
      cache-max-size-buffering: 1048576
      cleanup.on-shutdown: ${CLEANUP_ON_SHUTDOWN:false}
      properties:
        max:
          poll:
            records: 50

这有点不直观，但它确实有效。希望这对以后的人有帮助！

Spring 从 application.yaml 启动 Kafka StreamsConfig 或 ConsumerConfig 不适用

Spring Boot Kafka StreamsConfig or ConsumerConfig from application.yaml not applying

java

spring-boot

apache-kafka

spring-kafka