3

我们正在使用 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”。这怎么可能发生?

我想使用自定义状态存储的原因是,我(真的很容易)无法为应用程序实例提供专用硬盘。为了快速启动应用程序,我希望避免在每次启动应用程序时处理完整的变更日志(最好每天进行几次,目前需要一个多小时)。所以现在最后一个问题:

  • 使用自定义外部状态存储时,我能否在应用程序重新启动时恢复到最后一个状态?

[1] https://spring.io/blog/2019/12/09/stream-processing-with-spring-cloud-stream-and-apache-kafka-streams-part-6-state-stores-and-interactive -查询

4

1 回答 1

1

您正在使用Materialized.as(java.lang.String storeName)它将创建(实现)StateStore具有给定名称的 a (此处为“redis-store”)。另一方面,builder(RedisKeyValueStoreBytes redisKeyValueStoreBytes)您正在创建另一个 StateStore具有相同名称的 springframework 可能会自动将其添加到拓扑中,这样您就会收到“store is already added”错误。

q1:您可以在聚合中使用自定义状态存储;与Materialized.as(KeyValueBytesStoreSupplier 供应商)一起使用

q2:也可以将 aStateStore与转换器或自定义处理器一起用于交互式查询;也可以使用全局 StateStore访问整个主题,而不是仅分配分区的KafkaStreams实例(请参阅addGlobalStoreglobalTable

q3:我猜你没有(手动)用拓扑注册状态存储;请参阅Topology.addStateStore(StoreBuilder<?> storeBuilder, java.lang.String... processorNames)连接处理器和状态存储

q4:是的,状态存储是从变更日志主题加载的(使用优化时可能是原始主题)

于 2020-08-18T08:04:15.713 回答