application.yaml 中的 Spring Boot Kafka StreamsConfig 或 ConsumerConfig 不适用

Question

我有一个非常简单的带有 KTable 的 spring boot 项目，我想在 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)

ConsumerConfig：

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

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

@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

所以我在 如何在属性文件中正确外部化 spring-boot kafka-streams 配置？。

显然，使用 KStream 时，消费者和生产者配置与流配置完全分离。要为 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

这有点不直观，但它有效。希望这可以帮助将来的人！