我们正在使用 spring cloud stream Hoxton RC7 项目中包含的 Kafka-streams（因此使用提供的 Kafka-streams 和 Kafka-client 版本 [2.3.1]）
ext {
set('springCloudVersion', 'Hoxton.SR7')
}
...
dependencies {
// spring cloud stream
implementation 'org.springframework.cloud:spring-cloud-stream-binder-kafka-streams'
implementation 'org.springframework.cloud:spring-cloud-stream-binder-kafka'
implementation("org.springframework.cloud:spring-cloud-stream")
// redis
implementation 'io.lettuce:lettuce-core'
implementation 'org.springframework.data:spring-data-redis'
testCompile 'it.ozimov:embedded-redis:0.7.2'
...
我们已经实现了一个 kstreams 应用程序
@Bean
public Consumer<KStream<String, IncomingEvent>> process() {
return input -> {
我们在其中进行一些聚合，例如：
.aggregate(Foo::new, (key, value1, aggregate) ->
(aggregate == null || aggregate.getLastModified() == null || this.mustProcess(key, value1))
? value1
: aggregate,
materialized
)
现在物化应该是一个自定义的外部状态存储（Redis）：
Materialized<String, Foo, KeyValueStore<Bytes, byte[]>> materialized =
Materialized.as("redis-store");
由 StoreBuilder Bean 提供：
@Bean
public StoreBuilder<KeyValueStore<String, Foo>> builder(RedisKeyValueStoreBytes redisKeyValueStoreBytes){
return Stores.keyValueStoreBuilder(supplier(redisKeyValueStoreBytes),
new Serdes.StringSerde(),
new SomeFooSerde());
}
public static KeyValueBytesStoreSupplier supplier(RedisKeyValueStoreBytes redisKeyValueStoreBytes) {
return new KeyValueBytesStoreSupplier() {
@Override
public String name() {
return "redis-store";
}
@Override
public KeyValueStore<Bytes, byte[]> get() {
return redisKeyValueStoreBytes;
}
@Override
public String metricsScope() {
return "redis-session-state";
}
};
}
我现在使用 EmbeddedKafka 测试应用程序：
@ActiveProfiles("test")
@RunWith(SpringRunner.class)
@DirtiesContext(classMode = DirtiesContext.ClassMode.BEFORE_EACH_TEST_METHOD)
@SpringBootTest(classes = {TestConfigurationTests.class})
@EmbeddedKafka(count = 3, ports = {29901, 29902, 29903}, zookeeperPort = 33991)
public class TestKafkaIntegration {
我尝试访问状态存储并查询添加的项目：
ReadOnlyKeyValueStore<String, Foo> queryableStore = interactiveQueryService.getQueryableStore(
"redis-store", QueryableStoreTypes.keyValueStore());
return queryableStore;
但是当我运行我的测试时，我收到一个错误：
Caused by: org.springframework.kafka.KafkaException: Could not start stream: ; nested exception is org.springframework.kafka.KafkaException: Could not start stream: ; nested exception is org.apache.kafka.streams.errors.TopologyException: Invalid topology: StateStore redis-store is already added.
几个问题：
- [1] 解释的使用自定义状态存储的示例在处理器中使用它。这是否自动意味着，我无法在聚合中使用自定义状态存储？
- 如果无法在聚合中使用它，那么使用自定义状态存储有什么意义呢？
- 当我为 kstreams 稍微更改上面的代码并定义一个处理器而不是在聚合方法中使用物化时，错误会发生变化，然后它会在尝试执行 getQueryableStore 时抱怨缺少状态“redis-store”存储。但我确实可以看到，addStateStoreBeans 注册了“redis-store”。这怎么可能发生？
我想使用自定义状态存储的原因是，我（真的很容易）无法为应用程序实例提供专用硬盘。为了快速启动应用程序，我希望避免在每次启动应用程序时处理完整的变更日志（最好每天进行几次，目前需要一个多小时）。所以现在最后一个问题：
- 使用自定义外部状态存储时，我能否在应用程序重新启动时恢复到最后一个状态？