具有不同@StreamListener 的两个实例之间的嵌入式 Kafka 迁移状态存储答案

【问题标题】：Embedded Kafka migration State Store between two instances with different @StreamListener具有不同@StreamListener 的两个实例之间的嵌入式 Kafka 迁移状态存储
【发布时间】：2019-02-13 07:13:18
【问题描述】：

我有一个 SpringBoot 应用程序，其中有两个通过 Spring Cloud 映射的流处理器。每个处理器都有自己的 @StreamListener 用于不同的主题。一个处理器将聚合数据写入 quarable 状态存储。通过我的@Service（服务从状态存储获取聚合数据）获取数据时，我在单元测试中遇到了这个问题。由于某种原因不时捕获异常：

org.apache.kafka.streams.errors.InvalidStateStoreException: the state store, recently-played-store, may have migrated to another instance. at org.apache.kafka.streams.state.internals.QueryableStoreProvider.getStore(QueryableStoreProvider.java:60) at org.apache.kafka.streams.KafkaStreams.store(KafkaStreams.java:1043) at org.springframework.cloud.stream.binder.kafka.streams.QueryableStoreRegistry.getQueryableStoreType(QueryableStoreRegistry.java:47)

当我从另一个处理器中删除 StreamListener 时，一切正常且稳定。

如何使用适当的处理器为确切的实例绑定状态存储？

【问题讨论】：

标签： apache-kafka spring-cloud apache-kafka-streams spring-kafka embedded-kafka

【解决方案1】：

我找到了解决问题的方法，也许它会帮助其他人。我使用的不是最新版本的 spring-cloud-stream-binder-kafka-streams。我的版本是 2.0.0，它有一个错误 https://github.com/spring-cloud/spring-cloud-stream-binder-kafka/issues/366 。仅在 2.0.1 版本中修复

【讨论】：