【发布时间】：2020-09-18 01:52:41
【问题描述】：
我有一个基于 Spring Boot 的 Spring Cloud Stream Kafka Streams Binder 应用程序。 它定义了一个拓扑，其中包含以下部分：
绿色数字表示通过 Spring Cloud Stream Kafka Streams binder 绑定的各个处理器定义的拓扑传递的消息数量，以下是各自的属性：
spring.cloud.stream.bindings:
...
hint1Stream-out-0:
destination: hints
realityStream-out-0:
destination: hints
countStream-in-0:
destination: hints
我正在计算每个处理器使用peek() 方法产生/消耗的消息，如下所示：
return stream -> {
stream
.peek((k, v)-> input0count.incrementAndGet())
...
.peek((k, v)-> output0count.incrementAndGet())
};
我正在使用具有几乎默认设置的嵌入式 Kafka 从单元测试开始我的应用程序：
@RunWith(SpringRunner.class)
@SpringBootTest(
properties = "spring.cloud.stream.kafka.binder.brokers=${spring.embedded.kafka.brokers}"
)
@EmbeddedKafka(partitions = 1,
topics = {
...
TOPIC_HINTS
}
)
public class MyApplicationTests {
...
在我的测试中，我等待了足够长的时间，直到所有发布的测试消息都到达 countStream：
CountDownLatch latch = new CountDownLatch(1);
...
publishFromCsv(...)
...
latch.await(30, TimeUnit.SECONDS);
logCounters();
如您所见，放入“hints”主题的消息总和与“counterStream”端的消息计数不匹配：1309 + 2589 != 3786
我可能缺少一些 Kafka 或 Kafka Streams 设置来刷新每批？也许我的自定义 TimestampExtractor 会生成“太旧”的时间戳？ （我很确定它们不小于零）也许这与 Kafka 日志压缩有关？
这种不匹配的原因可能是什么？
更新
通过执行检查底层主题偏移量
kafka-run-class kafka.tools.GetOffsetShell --broker-list localhost:60231 --topic hints
当测试等待超时时。
正如预期的那样，主题中的消息数等于两个输入流计数的总和。传递到 counterStream 输入的消息数量仍然比预期的少几十个。
正在使用的其他 Kafka 配置：
spring.cloud.stream.kafka.streams:
configuration:
schema.registry.url: mock://torpedo-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
对应于processing.guarantee = at_least_once。无法测试 processing.guarantee = exactly_once，因为这需要至少有 3 个可用代理的集群。
同时设置：
spring.cloud.stream.kafka.binder.configuration:
auto.offset.reset: earliest
spring.cloud.stream.kafka.streams.binder.configuration:
auto.offset.reset: earliest
spring.cloud.stream.kafka.streams:
default:
consumer:
startOffset: earliest
spring.cloud.stream.bindings:
countStream-in-0:
destination: hints
consumer:
startOffset: earliest
concurrency: 1
没有帮助:(
帮助的是 stream.peak(..) 在 countStream 消费者中，例如：
@Bean
public Consumer<KStream<String, Hint>> countStream() {
return stream -> {
KStream<String, Hint> kstream = stream.peek((k, v) -> input0count.incrementAndGet());
};
}
在这种情况下，我立即开始在 countConsumer 端获得预期的消息数。
这意味着我的 Count Consumer 内部对行为有影响。
这是“不起作用”的完整版：
@Bean
public Consumer<KStream<String, Hint>> countStream() {
return stream -> {
KStream<String, Hint> kstream = stream.peek((k, v) -> notifyObservers(input0count.incrementAndGet()));
KStream<String, Hint> realityStream = kstream
.filter((key, hint) -> realityDetector.getName().equals(hint.getDetector()));
KStream<String, Hint> hintsStream = kstream
.filter((key, hint) -> !realityDetector.getName().equals(hint.getDetector()));
this.countsTable = kstream
.groupBy((key, hint) -> key.concat(":").concat(hint.getDetector()))
.count(Materialized
.as("countsTable"));
this.countsByActionTable = kstream
.groupBy((key, hint) -> key.concat(":")
.concat(hint.getDetector()).concat("|")
.concat(hint.getHint().toString()))
.count(Materialized
.as("countsByActionTable"));
this.countsByHintRealityTable = hintsStream
.join(realityStream,
(hint, real) -> {
hint.setReal(real.getHint());
return hint;
}, JoinWindows.of(countStreamProperties.getJoinWindowSize()))
.groupBy((key, hint) -> key.concat(":")
.concat(hint.getDetector()).concat("|")
.concat(hint.getHint().toString()).concat("-")
.concat(hint.getReal().toString())
)
.count(Materialized
.as("countsByHintRealityTable"));
};
}
我在那里将计数存储在几个 KTable 中。这就是 Counts Consumer 内部发生的事情：
更新 2
Count Consumer 的最后一部分显然导致了最初的意外行为：
this.countsByHintRealityTable = hintsStream
.join(realityStream,
(hint, real) -> {
hint.setReal(real.getHint());
return hint;
}, JoinWindows.of(countStreamProperties.getJoinWindowSize()))
.groupBy((key, hint) -> key.concat(":")
.concat(hint.getDetector()).concat("|")
.concat(hint.getHint().toString()).concat("-")
.concat(hint.getReal().toString())
)
.count(Materialized
.as("countsByHintRealityTable"));
没有它，消息计数会按预期匹配。
这样的下游代码如何影响 Consumer KStream 输入？
【问题讨论】：
-
想知道这是否是默认的 Kafka Streams 行为？我认为 Spring Cloud Stream binder 没有做任何事情来触发这种行为。可能想编写一个独立的 Kafka Streams 应用程序，看看你是否在那里看到同样的问题。
-
我打算检查基础主题的偏移量，但只会在几天内完成。
-
我想这真的取决于你的应用程序在做什么？有状态的操作员可能会“缓存和去重”对密钥的更新。消息可能格式不正确并被丢弃/跳过。一般来说，如果您没有在启用 EOS 的情况下运行并且没有正确的配置，则消息可能会被复制或丢弃。
-
我在分别计算输出和输入，我不希望有状态转换的影响在这里发挥作用。它也是 1 个嵌入式 Kafka 实例，所有主题都具有 1 个分区，为什么 Kafka 会因为 QoS 而放弃任何东西？但这是个好建议，谢谢，我会尝试更高的 QoS 级别。
-
格式错误的消息会导致日志中出现错误消息吗？图片上的所有消息都具有相同的密钥和 Avro 序列化内容。
标签： apache-kafka apache-kafka-streams spring-cloud-stream