Spring Cloud Stream 应用程序中的 timestampExtractorBeanName 设置不会覆盖默认值 [英] timestampExtractorBeanName setting in the Spring Cloud Stream application doesn't override default value
问题描述
我使用 Kafka Streams Binder 的 Spring Cloud Stream 应用程序具有以下属性:
I have the following properties for my Spring Cloud Stream application that uses Kafka Streams Binder:
spring.cloud.stream.bindings:
windowStream-in-0:
destination: input
windowStream-out-0:
destination: window1
hint1Stream-in-0:
destination: window1
hint1Stream-out-0:
destination: hints
realityStream-in-0:
destination: input
realityStream-in-1:
destination: window1
consumer:
timestampExtractorBeanName: anotherTimestampExtractor
realityStream-out-0:
destination: hints
countStream-in-0:
destination: hints
spring.cloud.stream.kafka.streams:
default:
consumer:
timestampExtractorBeanName: timestampExtractor
binder:
functions:
windowStream:
applicationId: mock-stream-window1
hint1Stream:
applicationId: mock-stream-hints
realityStream:
applicationId: mock-stream-reality
countStream:
applicationId: mock-stream-count
stateStoreRetry:
maxAttempts: 3
backOffInterval: 1000
configuration:
schema.registry.url: mock://mock-stream-registry
default.key.serde: org.apache.kafka.common.serialization.Serdes$StringSerde
default.value.serde: io.confluent.kafka.streams.serdes.avro.SpecificAvroSerde
commit.interval.ms: 100
我想要做的是对所有流使用timestampExtractor",除了一个叫做realityStream"的流.
What I am trying to do is to use the "timestampExtractor" for all streams except for one, called "realityStream".
为此,我将 spring.cloud.stream.kafka.streams.default.consumer.timestampExtractorBeanName
设置为 timestampExtractor
,然后尝试为现实流"通过设置 spring.cloud.stream.bindings.realityStream-in-1.consumer.timestampExtractorBeanName
For that I set the spring.cloud.stream.kafka.streams.default.consumer.timestampExtractorBeanName
to timestampExtractor
and then try to "override" it for the "realityStream" by setting spring.cloud.stream.bindings.realityStream-in-1.consumer.timestampExtractorBeanName
不幸的是,我的覆盖似乎不起作用,因为只有timestampExtractor"被调用,正如我在调试器(和我的测试结果)中看到的那样.
Unfortunately, it seems that my override doesn't work, since only "timestampExtractor" is getting called only, as I can see in the debugger (and in my test results).
我是否应用了错误的配置,或者有错误的期望?
Am I applying the configuration wrong, or have wrong expectation?
这是我的单个 Spring Cloud Streams 应用程序中的流图片:
Here's a picture of the streams in my single Spring Cloud Streams application:
(橙色圆圈是我想应用非默认时间戳提取器的地方)
(orange circle is the place where I want to apply non-default timestamp extractor)
推荐答案
覆盖在错误的地方;它需要在 ....kafka.streams.bindings.realityStream-in-1....
属性下.
The override is in the wrong place; it needs to be under the ....kafka.streams.bindings.realityStream-in-1....
property.
这是一个 kafka 特有的属性;您在通用绑定属性中拥有它(所有绑定器通用).
It's a kafka-specific property; you have it in the generic bindings properties (common to all binders).
这篇关于Spring Cloud Stream 应用程序中的 timestampExtractorBeanName 设置不会覆盖默认值的文章就介绍到这了,希望我们推荐的答案对大家有所帮助,也希望大家多多支持IT屋!