我们正在使用spring cloud stream hoxton rc7项目中包含的kafka流（因此使用提供的kafka流和kafka客户端版本[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";
        }
    };
}

我现在用一个嵌入的kafka测试应用程序：

@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稍微修改上面的代码并定义一个处理器而不是在aggregate方法中使用materialized时，错误发生了变化，然后它在尝试执行getqueryablestore时抱怨缺少状态“redis store”store。但实际上我可以看到，addstatestorebeans注册了“redis store”。怎么会这样？
我想使用自定义状态存储的原因是，我不能（很容易）为应用程序示例提供专用硬盘。为了快速启动应用程序，我希望避免在每次启动应用程序时处理完整的changelog（最好一天启动几次，目前需要一个多小时）。最后一个问题是：
使用自定义外部状态存储时，我是否能够恢复到应用程序重新启动时的最后一个状态？
[1] https://spring.io/blog/2019/12/09/stream-processing-with-spring-cloud-stream-and-apache-kafka-streams-part-6-state-stores-and-interactive-queries