如何通过YAML中的Spring Cloud Stream提供Kafka Streams属性？

Question

我想移动spring.kafka.streams.*在spring.cloud.stream下-这可能吗？我想到了流属性类似于消费者属性或生产者属性，但它不起作用。

spring:
  cloud:
    config:
      override-system-properties: false
      server:
        health:
          enabled: false
    stream:
      bindings:
        input_technischerplatz:
          destination: technischerplatz
        output_technischerplatz:
          destination: technischerplatz
      default:
        group: '${spring.application.name}'
        consumer:
          max-attempts: 5
      kafka:
        binder:
          auto-add-partitions: false
          auto-create-topics: false
          brokers: '${values.spring.kafka.bootstrap-servers}'
          configuration:
            header.mode: headers
          consumer-properties:
            allow.auto.create.topics: false
            auto.offset.reset: '${values.spring.kafka.consumer.auto-offset-reset}'
            enable.auto.commit: false
            isolation.level: read_committed
            max.poll.interval.ms: 300000
            max.poll.records: 100
            session.timeout.ms: 300000
          header-mapper-bean-name: defaultKafkaHeaderMapper
          producer-properties:
            acks: all
            key.serializer: org.apache.kafka.common.serialization.StringSerializer
            max.in.flight.requests.per.connection: 1
            max.block.ms: '${values.spring.kafka.producer.max-block-ms}'
            retries: 10
          required-acks: -1
  kafka:
    streams:
      applicationId: '${spring.application.name}_streams'
      properties:
        default.key.serde: org.apache.kafka.common.serialization.Serdes$StringSerde
        default.timestamp.extractor: org.apache.kafka.streams.processor.LogAndSkipOnInvalidTimestamp
        state.dir: '${values.spring.kafka.streams.properties.state.dir}'

Answer 1

您可以使用spring绑定streams属性。云按以下方式传输：

spring.cloud.stream.kafka.streams.binder.applicationId: my-application-id
spring.cloud.stream.kafka.streams.binder.configuration:
      default.key.serde: org.apache.kafka.common.serialization.Serdes$StringSerde
      default.value.serde: org.apache.kafka.common.serialization.Serdes$StringSerde

有关更多详细信息，您可以参考留档：

https://cloud.spring.io/spring-cloud-static/spring-cloud-stream-binder-kafka/3.0.0.M3/reference/html/spring-cloud-stream-binder-kafka.html#_kafka_streams_binder